Connection

The JSON-RPC 2.0 transport to the middleware. TrueNASWSConnection is what client.conn returns; it is rarely constructed directly, since TrueNASClient opens and authenticates one for you.

pytruenas.connection

A lean synchronous JSON-RPC 2.0 client for the TrueNAS middleware.

This is a deliberately small client covering exactly what pytruenas needs: open a websocket to the middleware, send a request, wait for the matching response, and surface errors as exceptions. Jobs are not tracked client-side -- the middleware's own core.job_wait method (a normal blocking call) is used for that. Event subscriptions (core.subscribe) ARE supported: see :meth:TrueNASWSConnection.subscribe and :class:Subscription -- a bounded queue drained on the caller's thread, with optional inline callbacks.

Three transports are supported by :class:TrueNASWSConnection:

• wss://host/api/current -- TLS websocket to a remote host;
• ws://host/api/current -- plain websocket to a remote host;
• ws+unix:///var/run/middleware/middlewared.sock -- the local middleware unix socket, used when running on the NAS (no network round trip, no auth needed for a root-owned socket).

The wire format is JSON-RPC 2.0 <https://www.jsonrpc.org/specification>_ with TrueNAS's extended-JSON value encoding (ejson below: datetime/date/ time/set/IP-interface round-tripping).

All annotations use quoted unions so the module imports on Python 3.9.

TrueNASWSConnection(uri=None, *, verify_ssl=True, call_timeout=CALL_TIMEOUT, py_exceptions=False, logger=None)

A synchronous JSON-RPC 2.0 client for the TrueNAS middleware.

uri may be a ws:///wss:// URL or a ws+unix:// unix-socket URL; None (or a bare ws+unix://) connects to the local middleware socket at :data:DEFAULT_UNIX_SOCKET.

The connection is a single blocking websocket. :meth:call sends one request and blocks on the matching response (by id); a background reader thread demultiplexes responses so concurrent calls from different threads are safe, and routes collection_update notifications to any :meth:subscribe sinks. Job tracking is intentionally omitted -- use the middleware's core.job_wait for jobs.

Source code in src/pytruenas/connection.py

def __init__(
    self,
    uri: "str | None" = None,
    *,
    verify_ssl: bool = True,
    call_timeout: float = CALL_TIMEOUT,
    py_exceptions: bool = False,
    logger: "_logging.Logger | _logging.LoggerAdapter | None" = None,
) -> None:
    if not uri or uri == _UNIX_PREFIX:
        uri = _UNIX_PREFIX + DEFAULT_UNIX_SOCKET
    self.uri = uri
    self.verify_ssl = verify_ssl
    self.call_timeout = call_timeout
    #: Where this connection's records go. A host passes its own
    #: name-bound logger, so "connection was closed" says *which* host
    #: closed even with several open at once; standalone use falls back to
    #: the module logger, which is what this class did unconditionally
    #: before (a module global cannot know its caller's target).
    self.logger = logger if logger is not None else _LOGGER
    # Accepted for parity with the upstream client; this lean client never
    # asks the server to pickle exceptions.
    self.py_exceptions = py_exceptions

    self._closed = _threading.Event()
    self._send_lock = _threading.Lock()
    self._pending: "dict[str, _Pending]" = {}
    self._pending_lock = _threading.Lock()
    # Event subscriptions, keyed by event name (the notification's
    # ``params.collection`` -- the routing key, verified against live 26.0).
    # A value is a LIST of sinks so two independent subscribers to the same
    # collection both receive it.
    self._subs: "dict[str, list[Subscription]]" = {}
    self._subs_lock = _threading.Lock()

    self._ws = self._connect()
    self._reader = _threading.Thread(target=self._read_loop, daemon=True)
    self._reader.start()

call_timeout = call_timeout instance-attribute

logger = logger if logger is not None else _LOGGER instance-attribute

py_exceptions = py_exceptions instance-attribute

uri = uri instance-attribute

verify_ssl = verify_ssl instance-attribute

__enter__()

Source code in src/pytruenas/connection.py

def __enter__(self) -> "TrueNASWSConnection":
    return self

__exit__(*exc)

Source code in src/pytruenas/connection.py

def __exit__(self, *exc) -> None:
    self.close()

call(method, *params, timeout=_UNSET, **_ignored)

Call method with params and return its result.

Blocks until the response arrives or timeout seconds elapse. The default sentinel uses :attr:call_timeout; an explicit timeout=None waits indefinitely (used by core.job_wait for long jobs). Compatibility keyword arguments (job, background, callback, …) accepted by the upstream client are ignored -- this client is synchronous and does not track jobs client-side. Any other unexpected keyword is logged at debug level rather than silently swallowed.

Raises :class:ValidationErrors/:class:ClientException on a server error, :class:CallTimeout on timeout, and :class:ClientException with errno=ECONNABORTED if the connection dropped.

Source code in src/pytruenas/connection.py

def call(
    self,
    method: str,
    *params,
    timeout: "float | None | _Unset" = _UNSET,
    **_ignored,
):
    """Call ``method`` with ``params`` and return its result.

    Blocks until the response arrives or ``timeout`` seconds elapse. The
    default sentinel uses :attr:`call_timeout`; an explicit ``timeout=None``
    waits **indefinitely** (used by ``core.job_wait`` for long jobs).
    Compatibility keyword arguments (``job``, ``background``, ``callback``,
    …) accepted by the upstream client are ignored -- this client is
    synchronous and does not track jobs client-side. Any *other* unexpected
    keyword is logged at debug level rather than silently swallowed.

    Raises :class:`ValidationErrors`/:class:`ClientException` on a server
    error, :class:`CallTimeout` on timeout, and :class:`ClientException`
    with ``errno=ECONNABORTED`` if the connection dropped.
    """
    unexpected = [k for k in _ignored if k not in _COMPAT_KWARGS]
    if unexpected:
        self.logger.debug(
            "call(%s): ignoring unexpected kwargs %s", method, unexpected
        )

    if self._closed.is_set():
        raise ClientException("Connection closed", _errno.ECONNABORTED)

    call_id = str(_uuid.uuid4())
    pending = _Pending()
    with self._pending_lock:
        self._pending[call_id] = pending

    request = {
        "jsonrpc": "2.0",
        "method": method,
        "id": call_id,
        "params": list(params),
    }
    try:
        with self._send_lock:
            self._ws.send(dumps(request))
    except Exception as exc:
        with self._pending_lock:
            self._pending.pop(call_id, None)
        raise ClientException(
            f"Failed to send request: {exc}", _errno.ECONNABORTED
        ) from exc

    wait_timeout = self.call_timeout if isinstance(timeout, _Unset) else timeout
    message = pending.wait(wait_timeout)
    if message is None:
        # Only reachable with a finite timeout; timeout=None waits forever.
        with self._pending_lock:
            self._pending.pop(call_id, None)
        raise CallTimeout()

    if "error" in message and message["error"] is not None:
        raise _parse_error(message["error"])
    return message.get("result")

close()

Close the websocket connection. Idempotent.

Source code in src/pytruenas/connection.py

def close(self) -> None:
    """Close the websocket connection. Idempotent."""
    if self._closed.is_set():
        return
    self._closed.set()
    # Wake any events() consumers; the server drops server-side
    # subscriptions when the socket closes.
    with self._subs_lock:
        sinks = [s for sinks in self._subs.values() for s in sinks]
        self._subs.clear()
    for sink in sinks:
        sink._close()
    try:
        self._ws.close()
    except Exception:
        pass

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

Subscribe to a middleware event and return a :class:Subscription.

event is a collection name (e.g. "alert.list", "reporting.realtime"). Consume via the returned subscription's :meth:~Subscription.events iterator and/or an optional callback (called inline on the reader thread -- keep it fast).

Registers the sink BEFORE issuing core.subscribe so no event can arrive between the server acknowledging and the sink existing; the sink is removed again if the subscribe call fails.

Source code in src/pytruenas/connection.py

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

    ``event`` is a collection name (e.g. ``"alert.list"``,
    ``"reporting.realtime"``). Consume via the returned subscription's
    :meth:`~Subscription.events` iterator and/or an optional ``callback``
    (called inline on the reader thread -- keep it fast).

    Registers the sink BEFORE issuing ``core.subscribe`` so no event can
    arrive between the server acknowledging and the sink existing; the sink
    is removed again if the subscribe call fails.
    """
    sub = Subscription(event, callback, maxsize, self)
    with self._subs_lock:
        self._subs.setdefault(event, []).append(sub)
    try:
        sub.id = self.call("core.subscribe", event)
    except Exception:
        with self._subs_lock:
            sinks = self._subs.get(event)
            if sinks and sub in sinks:
                sinks.remove(sub)
                if not sinks:
                    self._subs.pop(event, None)
        raise
    return sub

Subscription(event, callback, maxsize, client)

A live subscription to a middleware event (core.subscribe).

Obtain one from :meth:TrueNASWSConnection.subscribe. Consume events by iterating :meth:events (a bounded queue drains on the caller's thread -- no hidden threads, backpressure is visible) and/or by passing a callback at subscribe time (invoked inline on the reader thread -- keep it fast and non-blocking; a raising callback is logged and swallowed so it never kills the reader or other subscriptions).

Close with :meth:unsubscribe or a with block. When the queue fills (a slow consumer) the OLDEST event is dropped and :attr:dropped is incremented -- the reader never blocks.

Source code in src/pytruenas/connection.py

def __init__(
    self,
    event: str,
    callback: "_ty.Callable[[Event], object] | None",
    maxsize: int,
    client: "TrueNASWSConnection",
) -> None:
    self.event = event
    self.id: "str | None" = (
        None  # set by TrueNASWSConnection.subscribe after core.subscribe
    )
    self.dropped = 0
    self._queue: "_queue.Queue" = _queue.Queue(maxsize=maxsize)
    self._callback = callback
    self._client = client
    self._closed = False

__slots__ = ('event', 'id', 'dropped', '_queue', '_callback', '_client', '_closed') class-attribute instance-attribute

dropped = 0 instance-attribute

event = event instance-attribute

id = None instance-attribute

__enter__()

Source code in src/pytruenas/connection.py

def __enter__(self) -> "Subscription":
    return self

__exit__(*exc)

Source code in src/pytruenas/connection.py

def __exit__(self, *exc) -> None:
    self.unsubscribe()

events(timeout=None)

Yield events until the subscription (or the connection) closes.

Blocks on each event up to timeout seconds; timeout=None waits indefinitely. The iterator ends cleanly when :meth:unsubscribe is called or the connection drops.

Source code in src/pytruenas/connection.py

def events(self, timeout: "float | None" = None) -> "_ty.Iterator[Event]":
    """Yield events until the subscription (or the connection) closes.

    Blocks on each event up to ``timeout`` seconds; ``timeout=None`` waits
    indefinitely. The iterator ends cleanly when :meth:`unsubscribe` is
    called or the connection drops.
    """
    while True:
        if self._closed and self._queue.empty():
            return
        try:
            item = self._queue.get(timeout=timeout)
        except _queue.Empty:
            return  # timed out with nothing pending
        if item is _EVENT_CLOSED:
            return
        yield item

unsubscribe()

Cancel this subscription (idempotent; safe after the client closes).

Source code in src/pytruenas/connection.py

def unsubscribe(self) -> None:
    """Cancel this subscription (idempotent; safe after the client closes)."""
    self._client._unsubscribe(self)

Event

Bases: NamedTuple

One collection_update notification delivered to a subscription.

Mirrors the middleware wire shape verified against TrueNAS 26.0: {"method": "collection_update", "params": {"msg": ..., "collection": ..., "fields": {...}}}.

• collection -- the event name subscribed to (the routing key);
• msg -- "added" / "changed" / "removed";
• fields -- the record payload (may be None for some events);
• id -- the record id when the middleware includes one, else None.

collection instance-attribute

fields instance-attribute

id = None class-attribute instance-attribute

msg instance-attribute

ClientException(error, errno=None, trace=None, extra=None)

Bases: Exception

Any error surfaced by a :class:TrueNASWSConnection call or connection.

errno carries the middleware's error number when it maps to a POSIX errno (used by the filesystem layer to raise the right OSError); trace/extra carry the server-side traceback and per-field error detail when present.

Source code in src/pytruenas/connection.py

def __init__(
    self,
    error: str,
    errno: "int | None" = None,
    trace: "object | None" = None,
    extra: "list | None" = None,
) -> None:
    super().__init__(error)
    self.error = error
    self.errno = errno
    self.trace = trace
    self.extra = extra

errno = errno instance-attribute

error = error instance-attribute

extra = extra instance-attribute

trace = trace instance-attribute

__str__()

Source code in src/pytruenas/connection.py

def __str__(self) -> str:
    return self.error

ValidationErrors(errors)

Bases: ClientException

A collection of per-attribute validation errors from the server.

Each entry is (attribute, errmsg, errcode). The stringified form lists every attribute + message, matching how the middleware reports them.

Source code in src/pytruenas/connection.py

def __init__(self, errors: "_ty.Iterable") -> None:
    self.errors = [tuple(e) for e in errors]
    super().__init__(str(self))

errors = [(tuple(e)) for e in errors] instance-attribute

__str__()

Source code in src/pytruenas/connection.py

def __str__(self) -> str:
    msgs = []
    for attribute, errmsg, errcode in self.errors:
        name = _errno.errorcode.get(errcode, "EUNKNOWN")
        msgs.append(f"[{name}] {attribute or 'ALL'}: {errmsg}")
    return "\n".join(msgs)

CallTimeout()

Bases: ClientException

Raised when a call does not return within its timeout.

Source code in src/pytruenas/connection.py

def __init__(self) -> None:
    super().__init__("Call timeout", _errno.ETIMEDOUT)

dumps(obj, **kwargs)

Serialize obj to a JSON string, extended-type aware.

Source code in src/pytruenas/connection.py

def dumps(obj, **kwargs) -> str:
    """Serialize ``obj`` to a JSON string, extended-type aware."""
    return _json.dumps(obj, cls=_EJSONEncoder, **kwargs)

loads(data, **kwargs)

Deserialize a JSON string, decoding the middleware's extended types.

Source code in src/pytruenas/connection.py

def loads(data: "str | bytes | bytearray", **kwargs):
    """Deserialize a JSON string, decoding the middleware's extended types."""
    return _json.loads(data, object_hook=_object_hook, **kwargs)