Filesystem

pytruenas.fs

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

client.path(...) is the one constructor: hostctl's composite path, which routes each operation to this host's path providers in order and falls through to the next when one cannot serve the call. For a remote host that is sftp (over the host's own SSH connection) then tnasws; for the local socket it is a plain local path. client.path(..., backend="tnasws") pins the websocket leg.

This package holds the tnasws provider's path type, :class:~pytruenas.fs.tnasws.TrueNASWSPath, a real pathapi path, so every generic operation (read_bytes/walk/glob/copy/...) comes from pathapi.

__all__ = ['TrueNASWSPath', 'TrueNASWSBackend'] module-attribute

TrueNASWSBackend(client)

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

Mirrors the role of an SFTP backend (which holds an SSH 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

TrueNASWSPath

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._job(
        self._fs.setperm,
        {"path": self.path, "mode": oct(mode)[2:], "options": {"recursive": False}},
    )

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._job(
        self._fs.chown,
        {
            "path": self.path,
            "uid": uid,
            "gid": gid,
            "options": {"recursive": recursive, "traverse": traverse},
        },
    )

mkdir(mode=None, parents=False, exist_ok=False)

Create this directory; mode defaults to 0o755.

pathlib's 0o777 default is safe only because the kernel masks it with the process umask. The middleware applies the mode it is sent verbatim (measured live: a default mkdir() made a 0o777 directory), so the conventional 022 umask is applied here. An explicit mode is sent as given.

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

def mkdir(self, mode=None, parents=False, exist_ok=False):
    """Create this directory; ``mode`` defaults to ``0o755``.

    pathlib's ``0o777`` default is safe only because the kernel masks it
    with the process umask. The middleware applies the mode it is sent
    verbatim (measured live: a default ``mkdir()`` made a ``0o777``
    directory), so the conventional ``022`` umask is applied here. An
    explicit ``mode`` is sent as given.
    """
    mode = 0o777 & ~0o022 if mode is None else mode
    return super().mkdir(mode, parents=parents, exist_ok=exist_ok)

rmdir()

Remove this empty directory, as rmdir on the host does.

Raises FileNotFoundError, NotADirectoryError, or OSError with ENOTEMPTY when it has entries.

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

def rmdir(self):
    """Remove this empty directory, as ``rmdir`` on the host does.

    Raises ``FileNotFoundError``, ``NotADirectoryError``, or ``OSError``
    with ``ENOTEMPTY`` when it has entries.
    """
    done = self.backend.client.run(("rmdir", "--", self.path), check=False)
    if done.returncode == 0:
        return
    info = self._fs.stat(self.path, _ioerror=True)
    if (info.get("type") or "").upper() != "DIRECTORY":
        raise _error(_errno.ENOTDIR, self.path)
    error = _command_error(done, self.path)
    if error.errno == _errno.EIO and any(self._listdir()):
        raise _error(_errno.ENOTEMPTY, self.path)
    raise error

stat(*, follow_symlinks=True)

The path's status; follow_symlinks=False describes a link itself.

filesystem.stat describes a link itself (type: SYMLINK) and names its target in realpath, so following one is a second stat.

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

def stat(self, *, follow_symlinks=True) -> "_FileStat":
    """The path's status; ``follow_symlinks=False`` describes a link itself.

    ``filesystem.stat`` describes a link itself (``type: SYMLINK``) and
    names its target in ``realpath``, so following one is a second stat.
    """
    hint = self._pop_stat_hint()
    if hint is not None and not follow_symlinks:
        return hint
    info = self._fs.stat(self.path, _ioerror=True)
    if follow_symlinks and _entry_type(info) == "SYMLINK":
        target = info.get("realpath")
        if target and target != self.path:
            info = self._fs.stat(target, _ioerror=True)
    return _stat_from_info(info)

unlink(missing_ok=False)

Remove this file or link; a directory raises IsADirectoryError.

The filesystem.* API has no deletion, so this runs rm on the host. A failure is re-read with filesystem.stat to raise the OSError subclass pathlib would.

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

def unlink(self, missing_ok=False):
    """Remove this file or link; a directory raises ``IsADirectoryError``.

    The ``filesystem.*`` API has no deletion, so this runs ``rm`` on the
    host. A failure is re-read with ``filesystem.stat`` to raise the
    ``OSError`` subclass pathlib would.
    """
    done = self.backend.client.run(("rm", "--", self.path), check=False)
    if done.returncode == 0:
        return
    try:
        info = self._fs.stat(self.path, _ioerror=True)
    except FileNotFoundError:
        if missing_ok:
            return
        raise
    if (info.get("type") or "").upper() == "DIRECTORY":
        raise _error(_errno.EISDIR, self.path)
    raise _command_error(done, self.path)