Client

TrueNASClient is the client and the TrueNAS host in one class. Its hostctl half (run, path, capabilities, last_selection), the configuration and the transport providers are documented under Host.

pytruenas.TrueNASClient(config=None, **options)

TrueNASClient(
    config: "_ty.Optional[str]" = None,
    *,
    credentials: object = ...,
    username: "_ty.Optional[str]" = ...,
    password: "_ty.Optional[str]" = ...,
    otp: "_ty.Optional[str]" = ...,
    api_key: "_ty.Optional[str]" = ...,
    token: "_ty.Optional[str]" = ...,
    verify: "_ty.Union[bool, str, _os.PathLike[str], _ssl.SSLContext]" = ...,
    ssh: "_ty.Optional[SSHConfig]" = ...,
    version: str = ...,
    shell: "_ty.Optional[str]" = ...,
    known_hosts: object = ...,
    executor: "_ty.Optional[_ty.Union[_ty.Iterable[str], str]]" = ...,
    path: "_ty.Optional[_ty.Union[_ty.Iterable[str], str]]" = ...,
    autologin: bool = ...,
    timeout: "_ty.Union[float, None, Timeout]" = ...,
    executor_providers: "_ty.Optional[_ty.Iterable[ExecutorProvider]]" = ...,
    path_providers: "_ty.Optional[_ty.Iterable[PathProvider]]" = ...,
    info: "_ty.Optional[HostInfo]" = ...,
    initializer: object = ...
)

TrueNASClient(
    config: TrueNASConfig,
    *,
    executor_providers: "_ty.Optional[_ty.Iterable[ExecutorProvider]]" = ...,
    path_providers: "_ty.Optional[_ty.Iterable[PathProvider]]" = ...,
    shell: object = ...,
    info: "_ty.Optional[HostInfo]" = ...,
    initializer: object = ...
)

Bases: PosixHost, Generic[ApiVersion]

A TrueNAS middleware host: POSIX semantics over composed transports.

Everything generic -- run, path, spawn, info, connect, close, shell, capabilities, last_selection -- is inherited from :class:hostctl.host.PosixHost, which selects between the providers assembled below. What this class adds is only the part no other host has: the middleware JSON-RPC websocket and the API surface built on it.

Construct it from a connection string, a :class:TrueNASConfig, or nothing at all (the local middleware socket)::

TrueNASClient("wss://nas")
TrueNASClient("nas", credentials="1-...", executor=["ssh"])
TrueNASClient(TrueNASConfig.from_target("wss://nas"))
TrueNASClient()

A string is normalized by :func:_normalize_target, and the keyword options are those of :meth:TrueNASConfig.from_target.

Provider order depends on the target, and is overridable per selector with executor=/path=:

• local -- hostctl's stock local pair, and nothing else. Reaching this same machine over SSH, a PTY, or the filesystem API would be slower and strictly less capable.
• remote -- ssh then webshell for commands, sftp then tnasws for paths. SSH leads on capability; the websocket legs follow through hostctl's selector, which records a redacted trace of what was tried (host.last_selection).

Source code in src/pytruenas/host.py

def __init__(
    self,
    config: "_ty.Optional[_ty.Union[TrueNASConfig, str]]" = None,
    **options: object,
) -> None:
    # A connection string is the common case, so accept it directly rather
    # than making every caller reach for TrueNASConfig.from_target first.
    # hostctl's own `Host("uri")` shortcut cannot help here: its metaclass
    # only intercepts when `cls is Host`, so a subclass falls straight
    # through to normal construction.
    #
    # Keywords are split by destination: everything TrueNASConfig accepts
    # builds the config, and the rest (`info=`, `initializer=`, ...) goes
    # on to SystemHost. Splitting rather than guessing keeps a typo an
    # error from whichever layer owns the name.
    if config is None or isinstance(config, str):
        config_options = {
            key: options.pop(key) for key in list(options) if key in _CONFIG_OPTIONS
        }
        # Anything left that SystemHost does not take is a typo. Catch it
        # here rather than letting it reach SystemHost, which would raise a
        # TypeError naming an internal class -- unhelpful for a caller who
        # wrote `passwrd=` and needs to be told *that*.
        unknown = sorted(set(options) - _HOST_OPTIONS)
        if unknown:
            raise ValueError(
                f"unknown credential argument: {unknown[0]!r} "
                f"(configuration options: {', '.join(sorted(_CONFIG_OPTIONS))})"
            )
        config = TrueNASConfig.from_target(
            config, **_ty.cast(_ty.Any, config_options)
        )
    elif any(key in _CONFIG_OPTIONS for key in options):
        unexpected = ", ".join(
            sorted(key for key in options if key in _CONFIG_OPTIONS)
        )
        raise TypeError(
            "configuration options may not be combined with an existing "
            f"TrueNASConfig; pass them to from_target instead: {unexpected}"
        )
    self._config = config
    #: The live JSON-RPC connection, opened on first `.conn` access.
    self._conn: "_ty.Optional[_connection.TrueNASWSConnection]" = None
    # hostctl logs a provider fallback here too, once per skipped provider.
    _PosixHost.logger.fset(self, _resolve_logger(None, config.name))
    #: Guards opening, replacing and dropping `_conn`. Reentrant: logging
    #: in calls `auth.*` through `api`, which reads `conn` again.
    self._conn_lock = _threading.RLock()
    #: How the session last authenticated -- `login()`'s arguments -- so a
    #: reconnect repeats *that* login, not whatever the config would do.
    self._last_login: "_ty.Optional[tuple[object, dict]]" = None
    #: Closes the websocket if this host is collected without close().
    self._conn_finalizer: "_ty.Optional[_weakref.finalize]" = None
    #: One warning per host about credentials over plaintext ws://.
    self._plaintext_warned = False
    #: One warning per host when TLS verification is off.
    self._unverified_warned = False
    #: HTTP(S) session for the side channels; see `_http`.
    self._http_session: "_ty.Optional[_req.Session]" = None
    #: `{service: {method: is a job}}`, read once per service; see `_is_job`.
    self._job_flags: "dict[str, dict[str, bool]]" = {}

    super().__init__(config, **_ty.cast(_ty.Any, options))

JOB_DONE = ('SUCCESS', 'FAILED', 'ABORTED') class-attribute instance-attribute

api cached property

The root API namespace (host.api.<namespace>.<method>(...)).

Parameterise the client to type it: TrueNASClient[Current]("nas").api completes the generated API stubs.

config_type = TrueNASConfig class-attribute instance-attribute

conn property

The live JSON-RPC connection; opens on first access.

Logs in first when autologin is set (the default) and there is no live connection. Reconnects if the previous one closed; after a successful :meth:login, a reconnect repeats that login -- the same credentials and mechanism -- whatever autologin says.

Thread-safe: concurrent first use opens one connection, not one per thread.

logger property writable

The host's logger: every record it emits starts with [name].

Assign a logger name or a Logger to send the records elsewhere; it is bound to :attr:name the same way. A LoggerAdapter is used as it is. The JSON-RPC connection takes the logger set when it opens.

provider_resolvers = {'ssh': _ssh_pair, 'sftp': _ssh_pair, 'webshell': _webshell, 'tnasws': _tnasws, 'winrm': None} class-attribute instance-attribute

ssh property

The asyncssh connection run() uses (requires the ssh extra).

For the rare caller that needs the raw connection -- port forwarding, an SFTP client of its own. Ordinary command and path work should go through :meth:run and :meth:path, which pick a transport rather than assuming this one exists.

It is :attr:hostctl.transports.ssh.SSHExecutorProvider.connection: the first access connects (raising PermissionError, ConnectionError, TimeoutError or OSError), and the connection is bound to that provider's loop, hostctl's process-wide event-loop thread -- run its coroutines with asyncio.run_coroutine_threadsafe(coro, provider.loop).result(). :meth:close closes it. Raises RuntimeError when this host has no SSH executor.

close()

Close the transports, then the websocket.

Order matters: the tnasws path provider talks over the websocket, so it must be torn down before the connection it depends on.

Source code in src/pytruenas/host.py

def close(self) -> None:
    """Close the transports, then the websocket.

    Order matters: the ``tnasws`` path provider talks over the websocket,
    so it must be torn down before the connection it depends on.
    """
    finalizer, self._conn_finalizer = self._conn_finalizer, None
    if finalizer is not None:
        # Closing explicitly; nothing left for the collector to do.
        finalizer.detach()
    session, self._http_session = self._http_session, None
    if session is not None:
        session.close()
    try:
        super().close()
    finally:
        conn, self._conn = self._conn, None
        if conn is not None:
            try:
                conn.close()
            except Exception:
                # close() must be safe to call repeatedly and must not mask
                # an error raised by the provider teardown above.
                pass

connect()

Open the JSON-RPC connection, logging in when autologin is set.

This is what with TrueNASClient(...) does on entry. The command and path transports (SSH, the web shell, SFTP) connect when first used, so an API-only script opens no shell session and dials no SSH port. Raises what opening :attr:conn raises.

Source code in src/pytruenas/host.py

def connect(self) -> None:
    """Open the JSON-RPC connection, logging in when ``autologin`` is set.

    This is what ``with TrueNASClient(...)`` does on entry. The command and
    path transports (SSH, the web shell, SFTP) connect when first used, so
    an API-only script opens no shell session and dials no SSH port.
    Raises what opening :attr:`conn` raises.
    """
    self.conn

default_providers()

The provider names for this target when executor=/path= name none.

A local target is served entirely by hostctl's stock local pair: plain subprocess and plain local paths. Reaching this same machine through SSH, a PTY, or the filesystem API would be slower and strictly less capable, so no remote provider is offered.

Remotely, SSH leads on capability (separate streams, real stdin, a rich POSIX path surface) and the websocket legs follow: the web shell as a command channel for a host with no reachable SSH, and tnasws for paths. hostctl drops a default ssh/sftp when its ssh extra is missing; a name the caller gives is kept and fails at use.

Source code in src/pytruenas/host.py

def default_providers(self):
    """The provider names for this target when ``executor=``/``path=`` name none.

    A local target is served entirely by hostctl's stock local pair: plain
    subprocess and plain local paths. Reaching this same machine through
    SSH, a PTY, or the filesystem API would be slower and strictly less
    capable, so no remote provider is offered.

    Remotely, SSH leads on capability (separate streams, real stdin, a rich
    POSIX path surface) and the websocket legs follow: the web shell as a
    command channel for a host with no reachable SSH, and ``tnasws`` for
    paths. hostctl drops a default ``ssh``/``sftp`` when its ``ssh`` extra
    is missing; a name the caller gives is kept and fails at use.
    """
    config = self._config
    if config.is_local:
        return ("local",), ("local",)
    if config.ssh is not None:
        return ("ssh", "webshell"), ("sftp", "tnasws")
    return ("webshell",), ("tnasws",)

download(method, *args, filename=None, buffered=False, wait=True)

Call method for a download link and fetch it over HTTP(S).

With wait (the default) returns the content, once the job that produced it has finished: a failed job raises :class:~pytruenas.connection.JobFailed rather than returning what was sent before it failed. buffered=True waits before fetching; otherwise the job streams into the response and is checked after it. wait=False returns a :class:PendingDownload instead of the bytes: its job_id, and fetch() when they are wanted.

Source code in src/pytruenas/host.py

def download(
    self,
    method: str,
    *args,
    filename: "_ty.Optional[str]" = None,
    buffered=False,
    wait=True,
):
    """Call ``method`` for a download link and fetch it over HTTP(S).

    With ``wait`` (the default) returns the content, once the job that
    produced it has finished: a failed job raises
    :class:`~pytruenas.connection.JobFailed` rather than returning what
    was sent before it failed. ``buffered=True`` waits before fetching;
    otherwise the job streams into the response and is checked after it.
    ``wait=False`` returns a :class:`PendingDownload` instead of the bytes:
    its ``job_id``, and ``fetch()`` when they are wanted.
    """
    jobid, link = self.api.core.download(
        method, args, filename or "download", buffered
    )
    pending = PendingDownload(self, jobid, self._http_target(link).uri, buffered)
    if wait:
        return pending.fetch(callback=wait if callable(wait) else None)
    return pending

dump_api(*, cache=True, refresh=False, all_versions=False)

Return the middleware API definition (middlewared --dump-api).

The current API version only, pruned on the target before the transfer; all_versions=True keeps every version the appliance serves (current first), which is about five times the bytes on 26.0.

The dump is built and compressed on the target, then fetched over SFTP when this host has an SSH leg and in verified chunks otherwise. Its raw output cannot come back through run(): the dump is 24,103,000 bytes on 26.0, and the web shell's PTY loses the connection on that much output, on the very hosts (no SSH) that need it.

cache=True (the default) reads and writes the per-host, per-version cache under $PYTRUENAS_CACHE; building the dump costs about 75 seconds, which is not a price to pay twice. refresh=True re-fetches and overwrites it. See :mod:pytruenas.apicache.

Source code in src/pytruenas/host.py

def dump_api(
    self, *, cache: bool = True, refresh: bool = False, all_versions: bool = False
):
    """Return the middleware API definition (``middlewared --dump-api``).

    The current API version only, pruned on the target before the
    transfer; ``all_versions=True`` keeps every version the appliance
    serves (current first), which is about five times the bytes on 26.0.

    The dump is built and **compressed on the target**, then fetched over
    SFTP when this host has an SSH leg and in verified chunks otherwise.
    Its raw output cannot come back through ``run()``: the dump is
    24,103,000 bytes on 26.0, and the web shell's PTY loses the connection
    on that much output, on the very hosts (no SSH) that need it.

    ``cache=True`` (the default) reads and writes the per-host, per-version
    cache under ``$PYTRUENAS_CACHE``; building the dump costs about 75
    seconds, which is not a price to pay twice. ``refresh=True`` re-fetches
    and overwrites it. See :mod:`pytruenas.apicache`.
    """
    from . import apicache as _apicache

    if not cache:
        return _apicache.fetch(self, all_versions=all_versions)
    return _apicache.load(self, refresh=refresh, all_versions=all_versions)

install_sshcreds(name=None, private_key=None)

Provision an SSH keypair and wire it into this host's SSH transport.

Generates (or reuses) a SSH_KEY_PAIR keychain credential, installs the public half on root's authorized_keys, and stores the private half on :attr:TrueNASConfig.ssh as a real :class:hostctl.SSHConfig, as its client_keys; an ssh= the caller passed is copied, not changed.

Adding an SSH transport changes what this host can do, so the providers are rebuilt: a host that had no executor at all (remote, no SSH) gains one, and paths gain the richer SFTP leg.

Does not require the optional ssh extra: provisioning runs entirely over the middleware API and opens no SSH connection. The one exception is passing private_key= for a key the host does not already know, where the public half has to be derived locally (see :func:_public_key). Using the SSH transport this configures still needs the extra, as it always did.

Source code in src/pytruenas/host.py

def install_sshcreds(
    self, name: "_ty.Optional[str]" = None, private_key: "_ty.Optional[str]" = None
):
    """Provision an SSH keypair and wire it into this host's SSH transport.

    Generates (or reuses) a ``SSH_KEY_PAIR`` keychain credential, installs
    the public half on root's ``authorized_keys``, and stores the private
    half on :attr:`TrueNASConfig.ssh` as a real
    :class:`hostctl.SSHConfig`, as its ``client_keys``; an ``ssh=`` the
    caller passed is copied, not changed.

    Adding an SSH transport changes what this host can do, so the providers
    are rebuilt: a host that had no executor at all (remote, no SSH) gains
    one, and paths gain the richer SFTP leg.

    Does **not** require the optional ``ssh`` extra: provisioning runs
    entirely over the middleware API and opens no SSH connection. The one
    exception is passing ``private_key=`` for a key the host does not
    already know, where the public half has to be derived locally (see
    :func:`_public_key`). *Using* the SSH transport this configures still
    needs the extra, as it always did.
    """
    name = name or "pytruenas"
    keypair = self.api.keychaincredential._get(type="SSH_KEY_PAIR", name=name)
    # The middleware hands back BOTH halves on the paths it generates or
    # stores, so the public key is usually already known -- see `_public_key`
    # for why that matters.
    pubkey: "_ty.Optional[str]" = None
    if not keypair and not private_key:
        generated = self.api.keychaincredential.generate_ssh_key_pair()
        private_key = generated["private_key"]
        pubkey = generated.get("public_key")
    elif not private_key:
        attributes = keypair["attributes"]
        private_key = attributes["private_key"]
        pubkey = attributes.get("public_key")

    pubkey = (pubkey or "").strip() or _public_key(_ty.cast(str, private_key))
    keypair = self.api.keychaincredential._upsert(
        ("name", "type"),
        type="SSH_KEY_PAIR",
        name=name,
        attributes={"private_key": private_key, "public_key": pubkey},
    )
    root = self.api.user._get(username="root")
    authorized = (root.get("sshpubkey") or "").splitlines()
    if pubkey not in authorized:
        authorized.append(pubkey)
        # `("username",)` -- a one-item SEQUENCE, not the bare string. A
        # bare `str` selector is read as a record *id* (see
        # `DBAction.execute`); the tuple says "match on this field name",
        # which is what selecting root by username means.
        self.api.user._upsert(
            ("username",), username="root", sshpubkey="\n".join(authorized)
        )

    private_key = _ty.cast(str, keypair["attributes"]["private_key"])

    from hostctl import SSHConfig

    # A local target has no host to SSH *to*, and needs none -- commands
    # already run here. The keypair is still provisioned (it is installed
    # on root's authorized_keys, so other machines can use it), but there
    # is no leg to wire it into.
    if self._config.is_local:
        return private_key

    existing = self._config.ssh
    if existing is None:
        self._config.ssh = SSHConfig(
            host=self._config.host,
            username="root",
            client_keys=[private_key.encode()],
            verify=self._config.verify is not False,
            known_hosts=_ty.cast(_ty.Any, self._config.known_hosts),
        )
    elif not existing.password and not existing.client_keys:
        # A copy: the caller's `ssh=` object may serve other hosts too.
        self._config.ssh = _dataclasses.replace(
            existing, client_keys=[private_key.encode()]
        )
    else:
        # Explicit credentials win: a caller who configured their own auth
        # is not silently overridden by a provisioning call.
        return private_key

    # One resolution and one swap: the ssh and sftp legs share a
    # transport, and hostctl closes each replaced one (an open SSH
    # connection among them) once.
    executors, paths = self.resolve_providers()
    self.replace_providers(executor=executors, path=paths)
    return private_key

job_updates(job_id, *, timeout=None, poll=0.25, max_poll=2.0)

Yield job job_id's record each time its state or progress changes.

The record is core.get_jobs's: state (WAITING/RUNNING, then one of :attr:JOB_DONE), progress (percent, description), result, error... The first record is always yielded and the finished one is always the last, so a loop over this can drive a log line or a progress bar and then inspect the outcome::

for job in client.job_updates(job_id):
    print(job["progress"]["percent"], job["progress"]["description"])

Polls with a back-off from poll to max_poll seconds, reset whenever something changes. timeout=None waits indefinitely; otherwise :class:~pytruenas.connection.CallTimeout is raised (the job keeps running on the server). Failure is not raised here -- the final record says so; :meth:wait raises it.

Source code in src/pytruenas/host.py

def job_updates(
    self,
    job_id: int,
    *,
    timeout: "_ty.Optional[float]" = None,
    poll: float = 0.25,
    max_poll: float = 2.0,
) -> "_ty.Iterator[dict]":
    """Yield job ``job_id``'s record each time its state or progress changes.

    The record is ``core.get_jobs``'s: ``state`` (``WAITING``/``RUNNING``,
    then one of :attr:`JOB_DONE`), ``progress`` (``percent``,
    ``description``), ``result``, ``error``... The first record is always
    yielded and the finished one is always the last, so a loop over this
    can drive a log line or a progress bar and then inspect the outcome::

        for job in client.job_updates(job_id):
            print(job["progress"]["percent"], job["progress"]["description"])

    Polls with a back-off from ``poll`` to ``max_poll`` seconds, reset
    whenever something changes. ``timeout=None`` waits indefinitely;
    otherwise :class:`~pytruenas.connection.CallTimeout` is raised (the job
    keeps running on the server). Failure is not raised here -- the final
    record says so; :meth:`wait` raises it.
    """
    import time as _time

    deadline = None if timeout is None else _time.monotonic() + timeout
    backoff = _Backoff(poll, max_delay=max_poll)
    last = None
    while True:
        found = self.api.core.get_jobs([["id", "=", job_id]])
        if not found:
            raise _connection.ClientException(f"no such job: {job_id}")
        job = found[0]
        seen = (job.get("state"), _js.dumps(job.get("progress"), default=str))
        if seen != last:
            last = seen
            backoff.reset()
            yield job
        if job.get("state") in self.JOB_DONE:
            return
        if deadline is not None and _time.monotonic() + backoff.delay > deadline:
            raise _connection.CallTimeout()
        _time.sleep(backoff.delay)
        backoff.advance()

login(creds=None, *, login_ex=False, login_options=None, otp_provider=None)

Open a fresh connection and authenticate.

By default uses the legacy auth.login/login_with_* path. Pass login_ex=True for the modern mechanism, which supports 2FA via an OTP_REQUIRED continuation: the OTP comes from the credential's own otp if set, else from otp_provider(). A credential with no login_ex form (e.g. local-socket auth) falls back automatically.

Refused credentials raise :class:~pytruenas.auth.AuthenticationError on both paths, and the connection is closed rather than left open and unauthenticated. After a success, a reconnect repeats this login without the credential's one-time code, which is spent: on the login_ex path it asks otp_provider for a new one, and raises AuthenticationError("OTP_REQUIRED") without one.

Source code in src/pytruenas/host.py

def login(
    self,
    creds: "_ty.Optional[_auth.Credentials]" = None,
    *,
    login_ex: bool = False,
    login_options: "_ty.Optional[dict]" = None,
    otp_provider: "_ty.Optional[_ty.Callable[[], str]]" = None,
):
    """Open a fresh connection and authenticate.

    By default uses the legacy ``auth.login``/``login_with_*`` path. Pass
    ``login_ex=True`` for the modern mechanism, which supports 2FA via an
    ``OTP_REQUIRED`` continuation: the OTP comes from the credential's own
    ``otp`` if set, else from ``otp_provider()``. A credential with no
    login_ex form (e.g. local-socket auth) falls back automatically.

    Refused credentials raise :class:`~pytruenas.auth.AuthenticationError`
    on both paths, and the connection is closed rather than left open and
    unauthenticated. After a success, a reconnect repeats this login
    without the credential's one-time code, which is spent: on the
    ``login_ex`` path it asks ``otp_provider`` for a new one, and raises
    ``AuthenticationError("OTP_REQUIRED")`` without one.
    """
    options = {
        "login_ex": login_ex,
        "login_options": login_options,
        "otp_provider": otp_provider,
    }
    with self._conn_lock:
        if self._conn and not self._conn._closed.is_set():
            try:
                self._conn.close()
            except Exception:
                pass
        self._track(self._openwss())
        creds = creds or _ty.cast(_auth.Credentials, self._config.credentials)
        self._warn_if_plaintext(creds)
        try:
            if login_ex:
                result = creds.login_ex(
                    _ty.cast(_ty.Any, self),
                    login_options=login_options,
                    otp_provider=otp_provider,
                )
            else:
                result = creds.login(_ty.cast(_ty.Any, self))
        except BaseException:
            conn, self._conn = self._conn, None
            conn.close()
            raise
        self._last_login = (_without_otp(creds), options)
        return result

logout()

End the current session (auth.logout).

Source code in src/pytruenas/host.py

def logout(self) -> None:
    """End the current session (``auth.logout``)."""
    self.api.auth.logout()

me()

The current session's authenticated user (auth.me).

Source code in src/pytruenas/host.py

def me(self) -> dict:
    """The current session's authenticated user (``auth.me``)."""
    return _ty.cast(dict, self.api.auth.me())

path(*segments, backend=None)

A path on the target -- :meth:hostctl.host.Host.path, rooted at /.

On a remote target a relative path (and path() with no segments) names the same file on every leg: / joined with it. hostctl's SFTP leg reads a relative path against the SSH login directory, which the websocket leg cannot know, so left relative the same path would name different files depending on which leg serves it. A local target keeps local semantics (the process's working directory).

backend pins the path to that provider for every operation on it.

Source code in src/pytruenas/host.py

def path(self, *segments, backend=None):
    """A path on the target -- :meth:`hostctl.host.Host.path`, rooted at ``/``.

    On a remote target a relative path (and ``path()`` with no segments)
    names the same file on every leg: ``/`` joined with it. hostctl's SFTP
    leg reads a relative path against the SSH login directory, which the
    websocket leg cannot know, so left relative the same path would name
    different files depending on which leg serves it. A local target keeps
    local semantics (the process's working directory).

    ``backend`` pins the path to that provider for every operation on it.
    """
    if not self._config.is_local and (
        not segments or not str(segments[0]).startswith("/")
    ):
        segments = ("/", *segments)
    return super().path(*segments, backend=backend)

ping()

Round-trip the middleware (core.ping -> "pong").

Source code in src/pytruenas/host.py

def ping(self) -> str:
    """Round-trip the middleware (``core.ping`` -> ``"pong"``)."""
    return _ty.cast(str, self.api.core.ping())

upload(file, method, *params, token=None, wait=True)

Upload file via /_upload, then call method(*params) with it.

Source code in src/pytruenas/host.py

def upload(
    self,
    file: "_ty.Union[str, bytes]",
    method: str,
    *params,
    token=None,
    wait=True,
):
    """Upload ``file`` via ``/_upload``, then call ``method(*params)`` with it."""
    target = self._http_target("/_upload")
    data = {"method": method, "params": params}
    if isinstance(file, str):
        file = file.encode()

    if not token:
        # See the web shell: the server defaults match_origin to True and
        # this token is used once, by one HTTP request from this machine.
        token = self.api.auth.generate_token(5, {}, True, True)

    # An upload the middleware refuses answers with an HTTP error and an
    # HTML/JSON body; without the status check that surfaced as a KeyError
    # or a JSON decode error, naming neither the status nor the reason.
    resp = self._side_request(
        "POST",
        target.uri,
        headers={"Authorization": f"Token {token}"},
        files={"data": _js.dumps(data).encode(), "file": file},
    )
    try:
        jobid = resp.json()["job_id"]
    except (ValueError, KeyError, TypeError):
        raise _connection.ClientException(
            f"upload to {target.uri} returned no job id "
            f"(HTTP {resp.status_code}): {resp.text[:200]}"
        ) from None
    if wait:
        self.wait(jobid, callback=wait if callable(wait) else None)
    return jobid

wait(job_id, *, callback=None, timeout=None)

Block until middleware job job_id finishes; return its result.

callback(job) is called with the job record every time its state or progress changes (see :meth:job_updates) -- enough for a log line or a progress bar. SUCCESS returns the job's result; FAILED/ABORTED raise :class:~pytruenas.connection.JobFailed carrying the record.

Not core.job_wait: that method is itself a job, so calling it returns a NEW job id immediately instead of blocking (measured on 26.0.0-BETA.1: 0.4 s) -- every "wait" built on it returned early.

Source code in src/pytruenas/host.py

def wait(
    self,
    job_id: int,
    *,
    callback: "_ty.Optional[_ty.Callable[[dict], object]]" = None,
    timeout: "_ty.Optional[float]" = None,
) -> object:
    """Block until middleware job ``job_id`` finishes; return its result.

    ``callback(job)`` is called with the job record every time its state or
    progress changes (see :meth:`job_updates`) -- enough for a log line or
    a progress bar. ``SUCCESS`` returns the job's ``result``;
    ``FAILED``/``ABORTED`` raise :class:`~pytruenas.connection.JobFailed`
    carrying the record.

    Not ``core.job_wait``: that method is itself a job, so calling it
    returns a NEW job id immediately instead of blocking (measured on
    26.0.0-BETA.1: 0.4 s) -- every "wait" built on it returned early.
    """
    job: dict = {}
    for job in self.job_updates(job_id, timeout=timeout):
        if callback is not None:
            callback(job)
    if job.get("state") != "SUCCESS":
        raise _connection.JobFailed(job)
    return job.get("result")