Namespace

pytruenas.namespace.Namespace(client, *name)

Source code in src/pytruenas/namespace.py

def __init__(self, client: "TrueNASClient", *name: str) -> None:
    self._client = client
    self._namespace = ".".join([n for n in name if n])
    # Per-instance child cache. Using functools.cache on the methods keyed
    # the cache on ``self``, pinning every Namespace for the process
    # lifetime (uncollectable) -- fine for the CLI, a leak for a long-lived
    # embedding client. A plain dict is dropped with the instance.
    self._children: "dict[str, Namespace]" = {}

__call__(*args, _tries=1, _method=None, _ioerror=False, _filetransfer=False, _timeout=_UNSET, **kwds)

Invoke this namespace's middleware method.

_tries is the number of reconnect retries after a dropped connection (ECONNABORTED); with the default 1 the call is attempted up to twice. _timeout is the per-call timeout in seconds; the default sentinel uses the client's configured timeout, None waits indefinitely (used by core.job_wait for long jobs). _method appends a leaf method name, _ioerror maps a middleware error to the matching OSError, and _filetransfer routes through upload/download.

Source code in src/pytruenas/namespace.py

def __call__(
    self,
    *args,
    _tries=1,
    _method: str | None = None,
    _ioerror=False,
    _filetransfer: bool | bytes = False,
    _timeout: "float | None | _Unset" = _UNSET,
    **kwds,
):
    """Invoke this namespace's middleware method.

    ``_tries`` is the number of *reconnect retries* after a dropped
    connection (``ECONNABORTED``); with the default 1 the call is attempted
    up to twice. ``_timeout`` is the per-call timeout in seconds; the
    default sentinel uses the client's configured timeout, ``None`` waits
    indefinitely (used by ``core.job_wait`` for long jobs). ``_method``
    appends a leaf method name, ``_ioerror`` maps a middleware error to the
    matching ``OSError``, and ``_filetransfer`` routes through
    upload/download.
    """
    method = self._namespace
    if _method:
        method = f"{method}.{_method}"
    # _ioerror is only meaningful for the direct-call path below; the
    # transfer helpers don't accept it (it would leak into their **kwargs
    # and on to generate_token/core.download), so it is intentionally not
    # forwarded here.
    if _filetransfer is True:
        return self._client.download(method, *args, **kwds)
    elif _ioutils.isbytelike(_filetransfer) or hasattr(_filetransfer, "read"):
        return self._client.upload(
            _ty.cast(bytes, _filetransfer), method, *args, **kwds
        )
    elif _filetransfer:
        raise TypeError(_filetransfer)

    if _timeout is not _UNSET:
        kwds["timeout"] = _timeout

    # One initial attempt plus `_tries` reconnect retries after an aborted
    # connection. The loop always ends in a return or a raise -- it must
    # never fall through to None (a None here reads as "no result" and, via
    # _get, silently turns an upsert into a create).
    attempts = max(0, _tries) + 1
    last_exc: "_connection.ClientException | None" = None
    for attempt in range(attempts):
        try:
            self._client.logger.trace(  # type: ignore
                f"Calling method: {method} args: {args}"
            )
            return self._client.conn.call(method, *args, **kwds)
        except _connection.ClientException as e:
            last_exc = e
            if e.errno == _errno.ECONNABORTED and attempt < attempts - 1:
                self._client.logger.warning(
                    "Websocket connection was closed, trying again with new connection"
                )
                # Drop the dead connection on the CLIENT so the next
                # `conn` access reconnects (setting it on self, a
                # Namespace, was a no-op that only worked by accident).
                self._client._conn = None
                _time.sleep(1)
                continue
            raise (ioerror(e) if _ioerror else e) from None
    # Unreachable in practice (the loop returns or raises), but guarantees
    # we never return None on a connection error.
    raise ioerror(last_exc) if _ioerror else last_exc  # type: ignore[misc]

__getattr__(name)

Source code in src/pytruenas/namespace.py

def __getattr__(self, name: str) -> "Namespace": ...

__getitem__(name)

Source code in src/pytruenas/namespace.py

def __getitem__(self, name: str) -> "Namespace":
    child = self._children.get(name)
    if child is None:
        child = self._children[name] = Namespace(
            self._client, self._namespace, name
        )
    return child

__repr__()

Source code in src/pytruenas/namespace.py

def __repr__(self) -> str:  # type: ignore
    return f"{self.__class__.__name__}({self._client._api}/{self._namespace.replace('.','/')})"

__str__()

Source code in src/pytruenas/namespace.py

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

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

Subscribe to this namespace's collection events.

client.api.alert.list.subscribe() subscribes to the alert.list event -- the event name defaults to this namespace's dotted path. Returns a :class:~pytruenas.connection.Subscription; consume it via its events() iterator and/or the optional __callback (invoked inline on the reader thread -- keep it fast).

Pass event= to subscribe to a different event name than the namespace path (e.g. from client.api root). This is a real method, so it shadows any middleware method literally named subscribe; reach such a method via ns(_method="subscribe", ...) if ever needed.

Source code in src/pytruenas/namespace.py

def subscribe(
    self,
    __callback: "_ty.Callable[[_connection.Event], object] | None" = None,
    *,
    event: "str | None" = None,
    maxsize: int = _connection.DEFAULT_EVENT_QUEUE_SIZE,
) -> "_connection.Subscription":
    """Subscribe to this namespace's collection events.

    ``client.api.alert.list.subscribe()`` subscribes to the ``alert.list``
    event -- the event name defaults to this namespace's dotted path.
    Returns a :class:`~pytruenas.connection.Subscription`; consume it via its
    ``events()`` iterator and/or the optional ``__callback`` (invoked inline
    on the reader thread -- keep it fast).

    Pass ``event=`` to subscribe to a different event name than the
    namespace path (e.g. from ``client.api`` root). This is a real method,
    so it shadows any middleware method literally named ``subscribe``; reach
    such a method via ``ns(_method="subscribe", ...)`` if ever needed.
    """
    name = event or self._namespace
    if not name:
        raise ValueError(
            "no event name: call subscribe() on a namespace "
            "(client.api.alert.list.subscribe()) or pass event=..."
        )
    return self._client.conn.subscribe(name, __callback, maxsize=maxsize)

pytruenas.namespace.DbAction

Bases: str, Enum

The database mutation an _upsert/_update/_create resolves to.

A str enum (StrEnum is 3.11+, so this uses the 3.9-compatible (str, Enum) form) whose members double as the action name.

CREATE = 'create' class-attribute instance-attribute

UPDATE = 'update' class-attribute instance-attribute

UPSERT = 'upsert' class-attribute instance-attribute

execute(__action, __namespace, __selector=None, *__opts, **__fields)

Source code in src/pytruenas/namespace.py

def execute(
    __action,
    __namespace: "Namespace",
    __selector: _DBSelector = None,
    *__opts: dict | _q.Option | tuple[str, object],
    **__fields,
) -> _T:
    opts = _q.Option.options(*__opts)
    idkey = opts.get("idkey") or "id"
    callback = opts.get("callback") or None
    fields = {name: val for name, val in __fields.items() if val is not _q.EXCLUDE}

    if isinstance(__selector, str) or not isinstance(__selector, _ty.Iterable):
        _id = __selector
        selectors = {}
    else:
        _id = None
        selectors = (
            __selector
            if isinstance(__selector, _ty.Mapping)
            else {
                selector.removeprefix("!"): (
                    _q.EXCLUDE if selector.startswith("!") else fields[selector]
                )
                for selector in __selector
            }
        )

    if _id is None and selectors:
        current = __namespace._get(**selectors)
        if current:
            _id = _ty.cast(int, current[idkey])
    else:
        current = None

    force: bool = opts.get("force", False)
    action = None
    if _id is None and not selectors:
        if not force:
            result = _ty.cast(dict[str, object], __namespace.config())
            fields = _q.diff(result, fields)
        else:
            result = None
        if fields:
            result = __namespace.update(fields)
            action = DbAction.UPDATE
    elif _id not in (None, _q.EXCLUDE):
        if __action not in (DbAction.UPDATE, DbAction.UPSERT):
            raise FileExistsError(_id)
        exclude = (idkey, *(opts.get("update_exclude") or []))
        fields = {
            name: val
            for name, val in fields.items()
            if name not in exclude and selectors.get(name) not in (val, _q.EXCLUDE)
        }

        if not force:
            if not current:
                current = _ty.cast(dict[str, object], __namespace._get(_id))
            fields = _q.diff(current, fields)

        if fields:
            result = __namespace.update(_id, fields)
            action = DbAction.UPDATE
        else:
            result = None

        if result == _id:
            result = __namespace._get(_id)
        elif result is None:
            result = __namespace._get(_id) if not current else current
    else:
        if __action not in (DbAction.CREATE, DbAction.UPSERT):
            raise FileNotFoundError(selectors)
        exclude = (idkey, *(opts.get("create_exclude") or []))
        fields = {name: val for name, val in fields.items() if name not in exclude}
        try:
            result = __namespace.create(fields)
        except _connection.ValidationErrors as e:
            if "already exists" in e.error:
                raise FileExistsError(e.error)
            else:
                raise e
        action = DbAction.CREATE

    wait = opts.get("wait", True)
    if isinstance(result, int) and (wait is None or wait):
        result = __namespace._client.api.core.job_wait(
            result, job=True, _timeout=None
        )

    result = _ty.cast(_T, result)

    if callable(callback):
        callback(action, _id, result)

    return result