Filesystem

pytruenas.fs

Filesystem paths for a :class:~pytruenas.TrueNASClient.

client.path(...) returns a pathlib_next path:

• local client (running on the NAS) -> a plain :class:pathlib_next.LocalPath (no extra dependencies);
• remote client -> a :class:~pytruenas.fs.truenas.TruenasPath, which prefers SFTP and falls back to the middleware filesystem.* websocket API (:class:~pytruenas.fs.tnasws.TnasWsPath).

The old bespoke multi-backend Path proxy is gone; these are real pathlib_next path types, so every generic operation (read_bytes/walk/ glob/copy/...) comes from pathlib_next for free.

__all__ = ['LocalPath', 'TnasWsPath', 'TnasWsBackend', 'TruenasPath', 'path'] module-attribute

TnasWsBackend(client)

Backend state for :class:TnasWsPath: it just holds the client.

Mirrors the role of SftpBackend/AsyncsshSftpBackend (which hold an SSH/SFTP connection); here the "connection" is the middleware :class:~pytruenas.TrueNASClient, whose api.filesystem namespace does the actual work.

Source code in src/pytruenas/fs/tnasws.py

def __init__(self, client: "TrueNASClient") -> None:
    self.client = client

__slots__ = ('client',) class-attribute instance-attribute

client = client instance-attribute

TnasWsPath

Bases: UriPath

A path on a TrueNAS host, served by the middleware filesystem.* API.

__SCHEMES = (_SCHEME,) class-attribute instance-attribute

__slots__ = () class-attribute instance-attribute

chmod(mode, *, follow_symlinks=True)

Source code in src/pytruenas/fs/tnasws.py

def chmod(self, mode, *, follow_symlinks=True):
    # filesystem.setperm takes an octal-ish mode string.
    self._fs.setperm(
        {"path": self.path, "mode": oct(mode)[2:], "options": {"recursive": False}},
        _ioerror=True,
    )

chown(uid=None, gid=None, *, follow_symlinks=True, recursive=False, traverse=True)

Source code in src/pytruenas/fs/tnasws.py

def chown(
    self,
    uid=None,
    gid=None,
    *,
    follow_symlinks=True,
    recursive=False,
    traverse=True,
):
    if uid == -1:
        uid = None
    if gid == -1:
        gid = None
    if not follow_symlinks:
        raise NotImplementedError("chown(follow_symlinks=False)")
    self._fs.chown(
        {
            "path": self.path,
            "uid": uid,
            "gid": gid,
            "options": {"recursive": recursive, "traverse": traverse},
        },
        _ioerror=True,
    )

rmdir()

Source code in src/pytruenas/fs/tnasws.py

def rmdir(self):
    self.backend.client.run(("rmdir", self.path))

stat(*, follow_symlinks=True)

Source code in src/pytruenas/fs/tnasws.py

def stat(self, *, follow_symlinks=True) -> "_FileStat":
    hint = self._pop_stat_hint()
    if hint is not None and not follow_symlinks:
        return hint
    info = self._fs.stat(self.path, _ioerror=True)
    return _stat_from_info(info)

unlink(missing_ok=False)

Source code in src/pytruenas/fs/tnasws.py

def unlink(self, missing_ok=False):
    # The middleware has no filesystem.unlink; deletion is not part of the
    # filesystem.* API. Shell out on the host (the client runs commands
    # there). In a TruenasPath the SFTP leg handles this first when
    # available; this is the API-backend fallback.
    if missing_ok and not self.exists():
        return
    self.backend.client.run(("rm", "-f", self.path))

TruenasPath

Bases: TnasWsPath

Remote TrueNAS path: SFTP-preferred, websocket-filesystem.* fallback.

Subclasses :class:~pytruenas.fs.tnasws.TnasWsPath so the always-available websocket backend is the base behaviour; the SFTP-preferred operations (unlink/rmdir/rename/symlink_to/readlink) are overridden to try SFTP first. resolve is overridden too but never reaches SFTP -- SftpPath has no resolve -- so it always returns self. Carries the same TnasWsBackend (holding the client); the SFTP leg is built lazily from the host's SSH configuration (client.config.ssh, an :class:hostctl.host.SshConfig).

__SCHEMES = ('truenas',) class-attribute instance-attribute

__slots__ = () class-attribute instance-attribute

readlink()

Source code in src/pytruenas/fs/truenas.py

def readlink(self):
    try:
        link = self._try_sftp("readlink")
    except (_NoSftp, NotImplementedError):
        raise NotImplementedError("readlink requires the SFTP backend")
    return self.with_segments(_as_posix(link))

rename(target)

Source code in src/pytruenas/fs/truenas.py

def rename(self, target):
    try:
        return self._try_sftp("rename", _as_posix(target))
    except (_NoSftp, NotImplementedError):
        raise NotImplementedError("rename requires the SFTP backend")

resolve(strict=False)

Always returns self -- neither leg can resolve symlinks.

The SFTP attempt below is structural, not effective: pathlib_next's SftpPath implements no resolve, so _try_sftp raises NotImplementedError on every host and the fallback runs even when the SFTP leg is fully configured. The middleware has no realpath either, so returning the path unchanged is the best effort available -- matching a filesystem with no symlink resolution. The branch is kept so the operation starts working the day SftpPath grows a resolve.

Source code in src/pytruenas/fs/truenas.py

def resolve(self, strict=False):
    """Always returns ``self`` -- neither leg can resolve symlinks.

    The SFTP attempt below is structural, not effective: ``pathlib_next``'s
    ``SftpPath`` implements no ``resolve``, so ``_try_sftp`` raises
    ``NotImplementedError`` on every host and the fallback runs even when
    the SFTP leg is fully configured. The middleware has no ``realpath``
    either, so returning the path unchanged is the best effort available --
    matching a filesystem with no symlink resolution. The branch is kept so
    the operation starts working the day ``SftpPath`` grows a ``resolve``.
    """
    try:
        resolved = self._try_sftp("resolve", strict=strict)
    except (_NoSftp, NotImplementedError):
        return self
    return self.with_segments(_as_posix(resolved))

rmdir()

Source code in src/pytruenas/fs/truenas.py

def rmdir(self):
    try:
        return self._try_sftp("rmdir")
    except (_NoSftp, NotImplementedError):
        return super().rmdir()

symlink_to(target, target_is_directory=False, *, force=False, onremove=None)

Create a symlink, with pytruenas's force= convenience.

force removes a conflicting existing target first (a bool, a single file-type, or a set of file-types that may be replaced); onremove is a callback consulted before each removal. Not part of pathlib_next -- kept from the original pytruenas API. The link itself is created via the SFTP leg (filesystem.* has no symlink op), so a host with no SFTP leg raises NotImplementedError -- and raises it before force removes anything, so a call that cannot succeed leaves the existing target alone rather than deleting it and then failing.

Source code in src/pytruenas/fs/truenas.py

def symlink_to(
    self,
    target,
    target_is_directory=False,
    *,
    force: "bool | _FTYPE | _ty.Sequence[_FTYPE]" = False,
    onremove: "_ty.Callable[[_ty.Any, str], bool] | None" = None,
):
    """Create a symlink, with pytruenas's ``force=`` convenience.

    ``force`` removes a conflicting existing target first (a bool, a single
    file-type, or a set of file-types that may be replaced); ``onremove`` is a
    callback consulted before each removal. Not part of pathlib_next -- kept
    from the original pytruenas API. The link itself is created via the SFTP
    leg (``filesystem.*`` has no symlink op), so a host with no SFTP leg
    raises ``NotImplementedError`` -- and raises it *before* ``force``
    removes anything, so a call that cannot succeed leaves the existing
    target alone rather than deleting it and then failing.
    """
    # Resolve the creating leg up front. Building it is local (no network),
    # and doing it here is what keeps the force-removal below from running
    # on a host that has no way to create the replacement link.
    sftp = self._sftp()
    create = getattr(sftp, "symlink_to", None) if sftp is not None else None
    if create is None:
        raise NotImplementedError("symlink_to requires the SFTP backend")

    if force and (self.exists() or self.is_symlink()):
        onremove = onremove or (lambda _p, _t: True)
        if force is True:
            allowed = set(_ty.get_args(_FTYPE))
        elif isinstance(force, str):
            allowed = {force}
        else:
            allowed = set(force)
        if self.is_symlink():
            if "link" not in allowed:
                raise FileExistsError(self)
            if onremove(self, "link"):
                self.unlink()
        else:
            kind = (
                "directory"
                if self.is_dir()
                else "file" if self.is_file() else "unknown"
            )
            if kind not in allowed:
                raise FileExistsError(self)
            if onremove(self, kind):
                self.rm(recursive=True, missing_ok=True)
    return create(_as_posix(target), target_is_directory)

unlink(missing_ok=False)

Source code in src/pytruenas/fs/truenas.py

def unlink(self, missing_ok=False):
    try:
        return self._try_sftp("unlink", missing_ok=missing_ok)
    except (_NoSftp, NotImplementedError):
        return super().unlink(missing_ok=missing_ok)

path(client, *segments, backend=None)

Build the appropriate path type for client and segments.

backend forces a specific type: "local" -> :class:LocalPath, "ws"/"api" -> :class:TnasWsPath, "truenas"/"auto" -> :class:TruenasPath. Default (None/"auto"): LocalPath for a local client, otherwise TruenasPath.

Source code in src/pytruenas/fs/__init__.py

def path(client: "TrueNASClient", *segments, backend: "str | None" = None):
    """Build the appropriate path type for ``client`` and ``segments``.

    ``backend`` forces a specific type: ``"local"`` -> :class:`LocalPath`,
    ``"ws"``/``"api"`` -> :class:`TnasWsPath`, ``"truenas"``/``"auto"`` ->
    :class:`TruenasPath`. Default (``None``/``"auto"``): ``LocalPath`` for a local
    client, otherwise ``TruenasPath``.
    """
    backend = backend or "auto"
    posix = "/".join(str(s) for s in segments) if segments else "/"

    if backend == "local" or (backend == "auto" and _settings(client).is_local):
        return LocalPath(*segments) if segments else LocalPath("/")

    ws_backend = TnasWsBackend(client)
    if backend in ("ws", "api"):
        return TnasWsPath(_ws_uri(client, posix), backend=ws_backend)
    return TruenasPath(_truenas_uri(client, posix), backend=ws_backend)