Client

TrueNASClient and TrueNASHost are the same class. "Client" is the friendlier name at a call site; the class is documented under Host, where its hostctl half (run, path, capabilities, last_selection) lives.

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

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)::

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

A string accepts every form :class:~pytruenas.TrueNASClient does and takes the same keyword options as :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. This reproduces :class:~pytruenas.fs.truenas.TruenasPath's hand-rolled fallback through hostctl's selector, which additionally records a redacted trace of what was tried (host.last_selection).

Source code in src/pytruenas/host.py

def __init__(
    self,
    config: "TrueNASConfig | str | None" = None,
    credentials: object = None,
    *,
    client: object = None,
    **options: object,
) -> None:
    # `TrueNASClient(target, creds)` passed credentials positionally, and
    # that is the single most common call in the wild -- accept it.
    if credentials is not None:
        if "credentials" in options:
            raise TypeError("credentials given both positionally and by keyword")
        options["credentials"] = credentials
    # 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: "_connection.TrueNASWSConnection | None" = None
    self.logger = _resolve_logger(config.logger, config.name)
    # `client=` is accepted and ignored: the host *is* the client now.
    # Kept so existing callers (and tests that injected a stand-in) do not
    # break on an unexpected keyword.
    del client

    executors, paths = self._build_providers(config)
    super().__init__(
        config,
        executor_providers=executors,
        path_providers=paths,
        **_ty.cast(_ty.Any, options),
    )

api cached property

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

Parameterise the host to type it: TrueNASHost[Current]("nas").api completes exactly as the old TrueNASClient[Current] did.

client property

Deprecated alias for self.

The host is the client -- they were two objects forwarding halves of their surface to each other, which is now one class. Kept so host.client.api and similar keep working.

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.

logger = _resolve_logger(config.logger, config.name) instance-attribute

name property

This host's short label -- see :attr:TrueNASConfig.name.

ssh property

The underlying asyncssh connection (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.

websocket property

Former name of :attr:conn, kept because it is public API.

Defined as a property that reads self.conn rather than as websocket = conn: the latter makes two independent class attributes, so patching or overriding one would silently leave the other pointing at the original.

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.
    """
    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

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

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

Source code in src/pytruenas/host.py

def download(
    self,
    method: str,
    *args,
    filename: "str | None" = None,
    buffered=False,
    wait=True,
    **kwargs,
):
    """Call ``method`` for a download link and fetch it over HTTP(S)."""
    jobid, link = self.api.core.download(
        method, args, filename or "download", buffered, **kwargs
    )
    target = self._http_target(link)

    if wait:
        if buffered:
            self.api.core.job_wait(jobid, job=True, _timeout=None)
        resp = _req.get(target.uri, verify=self._config.sslverify)
        resp.raise_for_status()
        return resp.content
    return jobid

dump_api()

Run middlewared --dump-api on the target and parse the JSON.

Source code in src/pytruenas/host.py

def dump_api(self):
    """Run ``middlewared --dump-api`` on the target and parse the JSON."""
    import json

    from .models.apidump import Api

    api: Api = json.loads(
        self.run("middlewared --dump-api", capture_output=True).stdout
    )
    return api

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.host.SshConfig -- with client_keys as an actual field rather than the "client_keys|root" string that the pre-hostctl client packed into a username.

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.

Source code in src/pytruenas/host.py

def install_sshcreds(
    self, name: "str | None" = None, private_key: "str | None" = 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.host.SshConfig` -- with ``client_keys`` as an actual
    field rather than the ``"client_keys|root"`` string that the pre-hostctl
    client packed into a username.

    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.
    """
    name = name or "pytruenas"
    keypair = self.api.keychaincredential._get(type="SSH_KEY_PAIR", name=name)
    if not keypair and not private_key:
        private_key = self.api.keychaincredential.generate_ssh_key_pair()[
            "private_key"
        ]
    elif not private_key:
        private_key = keypair["attributes"]["private_key"]

    pubkey = (
        _asyncssh()
        .import_private_key(private_key)
        .export_public_key()
        .decode()
        .strip()
    )
    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)
        self.api.user._upsert(
            "username", username="root", sshpubkey="\n".join(authorized)
        )

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

    from hostctl.host 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()],
        )
    elif not existing.password and not existing.client_keys:
        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

    executors, paths = self._build_providers(self._config)
    from hostctl.provider import ProviderSelector

    self._executor_selector = ProviderSelector(executors)
    self._path_selector = ProviderSelector(paths)
    return private_key

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_token if set, else from otp_provider(). A credential with no login_ex form (e.g. local-socket auth) falls back automatically.

Source code in src/pytruenas/host.py

def login(
    self,
    creds: "_auth.Credentials | None" = None,
    *,
    login_ex: bool = False,
    login_options: "dict | None" = None,
    otp_provider: "_ty.Callable[[], str] | None" = 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_token`` if set, else from ``otp_provider()``. A credential with no
    login_ex form (e.g. local-socket auth) falls back automatically.
    """
    if self._conn and not self._conn._closed.is_set():
        try:
            self._conn.close()
        except Exception:
            pass
    self._conn = self._openwss()
    creds = creds or _ty.cast(_auth.Credentials, self._config.credentials)
    if login_ex:
        return creds.login_ex(
            _ty.cast(_ty.Any, self),
            login_options=login_options,
            otp_provider=otp_provider,
        )
    creds.login(_ty.cast(_ty.Any, self))

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())

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())

subscribe(event, callback=None, *, maxsize=_connection.DEFAULT_EVENT_QUEUE_SIZE)

Subscribe to a middleware event; return a Subscription.

host.subscribe("alert.list") is shorthand for host.api.alert.list.subscribe(). A subscription is bound to the current websocket and does not survive a reconnect -- the events() iterator ending is that signal.

Source code in src/pytruenas/host.py

def subscribe(
    self,
    event: str,
    callback: "_ty.Callable[..., object] | None" = None,
    *,
    maxsize: int = _connection.DEFAULT_EVENT_QUEUE_SIZE,
):
    """Subscribe to a middleware event; return a ``Subscription``.

    ``host.subscribe("alert.list")`` is shorthand for
    ``host.api.alert.list.subscribe()``. A subscription is bound to the
    current websocket and does **not** survive a reconnect -- the
    ``events()`` iterator ending is that signal.
    """
    return self.conn.subscribe(event, callback, maxsize=maxsize)

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

Upload file via /_upload, then call method with it.

Source code in src/pytruenas/host.py

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

    if not token:
        token = self.api.auth.generate_token(5, {}, False, **kwargs)

    resp = _req.post(
        target.uri,
        headers={"Authorization": f"Token {token}"},
        verify=self._config.sslverify,
        files={"data": _js.dumps(data).encode(), "file": file},
    )
    jobid = resp.json()["job_id"]
    if wait:
        self.api.core.job_wait(jobid, job=True, _timeout=None)
    return jobid