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/resolve) are overridden to try SFTP first. Carries the same TnasWsBackend (holding the client); the SFTP leg is built lazily from the client's shell/ssh configuration.

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

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

def resolve(self, strict=False):
    try:
        resolved = self._try_sftp("resolve", strict=strict)
    except (_NoSftp, NotImplementedError):
        # No SFTP: the middleware has no realpath; return self unchanged
        # (best effort, matching a filesystem with no symlink resolution).
        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).

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).
    """
    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)
    try:
        return self._try_sftp("symlink_to", _as_posix(target), target_is_directory)
    except (_NoSftp, NotImplementedError):
        raise NotImplementedError("symlink_to requires the SFTP backend")

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 client._api.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)