Skip to content

Protocols API

The small protocols Path is built from. A Path subclass implements the primitives (stat, _open, chmod, _chown) and inherits everything derived from them.

pathlib_next.protocols.fs

Chmod

Bases: Protocol

Composable protocol for objects supporting permission changes.

chmod(mode, *, follow_symlinks=True)

Change the permissions of the path, like os.chmod().

Extension: mode may be a str, which is parsed as octal ("0755", "755" and 0o755 all mean the same thing) -- see utils.as_mode() for why the base is never left implicit.

Every implementation normalizes its argument through utils.as_mode() on entry. Unlike symlink_to/chown, chmod is overridden directly by each backend (each has real per-scheme logic -- version shims, capability gates, a SITE CHMOD command), so there is no single wrapper to hang the conversion on; the shared helper is what keeps the base from drifting between them.

Source code in src/pathlib_next/protocols/fs.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
@_utils.notimplemented
def chmod(self, mode: int | str, *, follow_symlinks: bool = True):
    """
    Change the permissions of the path, like os.chmod().

    Extension: `mode` may be a `str`, which is parsed as **octal**
    (`"0755"`, `"755"` and `0o755` all mean the same thing) -- see
    `utils.as_mode()` for why the base is never left implicit.

    Every implementation normalizes its argument through
    `utils.as_mode()` on entry. Unlike `symlink_to`/`chown`, `chmod` is
    overridden directly by each backend (each has real per-scheme logic
    -- version shims, capability gates, a `SITE CHMOD` command), so
    there is no single wrapper to hang the conversion on; the shared
    helper is what keeps the base from drifting between them.
    """
    ...

chown(uid=None, gid=None, *, follow_symlinks=True)

Change the owner and/or group of the path.

Extension: pathlib.Path has owner()/group() readers but no writer (the stdlib equivalent is shutil.chown/os.chown), so ownership was the one attribute stat() could report -- via st_uid/st_gid -- and nothing could write back.

None (the default) leaves a field unchanged, which is what lets a caller set the group without knowing the owner. -1 is accepted as an alias for None, matching os.chown's own sentinel. An int is a uid/gid; a str is a user/group name, passed through for backends that can resolve one.

Backends implement _chown() and receive an already-canonical pair (see utils.as_owner()), so the "unchanged" semantics cannot drift between schemes.

Source code in src/pathlib_next/protocols/fs.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
def chown(
    self,
    uid: int | str | None = None,
    gid: int | str | None = None,
    *,
    follow_symlinks: bool = True,
) -> None:
    """Change the owner and/or group of the path.

    Extension: `pathlib.Path` has `owner()`/`group()` **readers** but no
    writer (the stdlib equivalent is `shutil.chown`/`os.chown`), so
    ownership was the one attribute `stat()` could report -- via
    `st_uid`/`st_gid` -- and nothing could write back.

    `None` (the default) leaves a field unchanged, which is what lets a
    caller set the group without knowing the owner. `-1` is accepted as
    an alias for `None`, matching `os.chown`'s own sentinel. An `int` is
    a uid/gid; a `str` is a user/group name, passed through for backends
    that can resolve one.

    Backends implement `_chown()` and receive an already-canonical pair
    (see `utils.as_owner()`), so the "unchanged" semantics cannot drift
    between schemes.
    """
    uid, gid = _utils.as_owner(uid, gid)
    if uid is None and gid is None:
        # Nothing to change -- don't spend a round trip (or make a
        # backend decide what an all-sentinel call means).
        return
    return self._chown(uid, gid, follow_symlinks=follow_symlinks)

lchmod(mode)

Like chmod(), except if the path points to a symlink, the symlink's permissions are changed, rather than its target's.

Source code in src/pathlib_next/protocols/fs.py 
134
135
136
137
138
139
def lchmod(self, mode: int | str):
    """
    Like chmod(), except if the path points to a symlink, the symlink's
    permissions are changed, rather than its target's.
    """
    self.chmod(mode, follow_symlinks=False)

FileStatLike

Bases: Protocol

Minimum properties stat like object should provide

Stat

Bases: Protocol

Any object that can implement Stat and utilities functions based on it

exists(*, follow_symlinks=True)

Whether this path exists.

Source code in src/pathlib_next/protocols/fs.py 
60
61
62
63
64
def exists(self, *, follow_symlinks=True):
    """
    Whether this path exists.
    """
    return self._st_mode(follow_symlinks=follow_symlinks) is not None

is_block_device()

Whether this path is a block device.

Source code in src/pathlib_next/protocols/fs.py 
86
87
88
89
90
def is_block_device(self):
    """
    Whether this path is a block device.
    """
    return _stat.S_ISBLK(self._st_mode() or 0)

is_char_device()

Whether this path is a character device.

Source code in src/pathlib_next/protocols/fs.py 
92
93
94
95
96
def is_char_device(self):
    """
    Whether this path is a character device.
    """
    return _stat.S_ISCHR(self._st_mode() or 0)

is_dir(*, follow_symlinks=True)

Whether this path is a directory. follow_symlinks=False reports a symlink to a directory as not one (3.13 parity).

Source code in src/pathlib_next/protocols/fs.py 
66
67
68
69
70
71
def is_dir(self, *, follow_symlinks=True):
    """
    Whether this path is a directory. `follow_symlinks=False` reports a
    symlink to a directory as not one (3.13 parity).
    """
    return _stat.S_ISDIR(self._st_mode(follow_symlinks=follow_symlinks) or 0)

is_fifo()

Whether this path is a FIFO.

Source code in src/pathlib_next/protocols/fs.py 
 98
 99
100
101
102
def is_fifo(self):
    """
    Whether this path is a FIFO.
    """
    return _stat.S_ISFIFO(self._st_mode() or 0)

is_file(*, follow_symlinks=True)

Whether this path is a regular file (also True for symlinks pointing to regular files unless follow_symlinks=False, 3.13 parity).

Source code in src/pathlib_next/protocols/fs.py 
73
74
75
76
77
78
def is_file(self, *, follow_symlinks=True):
    """
    Whether this path is a regular file (also True for symlinks pointing
    to regular files unless `follow_symlinks=False`, 3.13 parity).
    """
    return _stat.S_ISREG(self._st_mode(follow_symlinks=follow_symlinks) or 0)

is_socket()

Whether this path is a socket.

Source code in src/pathlib_next/protocols/fs.py 
104
105
106
107
108
def is_socket(self):
    """
    Whether this path is a socket.
    """
    return _stat.S_ISSOCK(self._st_mode() or 0)

Whether this path is a symbolic link.

Source code in src/pathlib_next/protocols/fs.py 
80
81
82
83
84
def is_symlink(self):
    """
    Whether this path is a symbolic link.
    """
    return _stat.S_ISLNK(self._st_mode(follow_symlinks=False) or 0)

lstat()

Like stat(), except if the path points to a symlink, the symlink's status information is returned, rather than its target's.

Source code in src/pathlib_next/protocols/fs.py 
34
35
36
37
38
39
def lstat(self) -> FileStatLike:
    """
    Like stat(), except if the path points to a symlink, the symlink's
    status information is returned, rather than its target's.
    """
    return self.stat(follow_symlinks=False)

pathlib_next.protocols.io

BinaryOpen

Bases: Protocol

Protocol for objects that support open->io.IoBase

copy(target, *, progress=None, chunk_size=_shutil.COPY_BUFSIZE)

Copy the binary content from this object to a target object.

progress, when given, is called after each chunk is written as progress(bytes_copied, total_size): bytes_copied increases monotonically and equals total_size (if known) after the final call; an empty file gets exactly one call, progress(0, total_size). total_size is this object's stat().st_size when self also implements the Stat protocol and stat() succeeds, otherwise None -- BinaryOpen alone has no size concept. chunk_size controls how many bytes are read per iteration (default: shutil.COPY_BUFSIZE). With progress=None (the default), behavior and bytes-on-wire are identical to before this was added (a plain shutil.copyfileobj).

Source code in src/pathlib_next/protocols/io.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
def copy(
    self,
    target: "BinaryOpen",
    *,
    progress: "_ty.Callable[[int, _ty.Optional[int]], None]" = None,
    chunk_size: int = _shutil.COPY_BUFSIZE,
):
    """Copy the binary content from this object to a target object.

    `progress`, when given, is called after each chunk is written as
    `progress(bytes_copied, total_size)`: `bytes_copied` increases
    monotonically and equals `total_size` (if known) after the final
    call; an empty file gets exactly one call, `progress(0, total_size)`.
    `total_size` is this object's `stat().st_size` when `self` also
    implements the `Stat` protocol and `stat()` succeeds,
    otherwise `None` -- `BinaryOpen` alone has no size concept.
    `chunk_size` controls how many bytes are read per iteration
    (default: `shutil.COPY_BUFSIZE`). With `progress=None` (the
    default), behavior and bytes-on-wire are identical to before this
    was added (a plain `shutil.copyfileobj`).
    """
    # The source is opened first: opening the target "wb" truncates or
    # creates it, and doing that before knowing the source is readable
    # turned a missing source into an emptied destination.
    with self.open("rb") as input, target.open("wb") as output:
        self._copy_stream(input, output, progress=progress, chunk_size=chunk_size)

open(mode='r', buffering=-1, encoding=None, errors=None, newline=None)

Open the a handle to an object that implement io.IOBase

The mode is validated like open() (ValueError for an invalid mode, or an encoding/errors/newline with a binary one), and _open() receives it canonical: one of "r"/"w"/"x"/"a", plus "+" if given -- never "b" or "t", so "rt"/"wt" reach every backend as "r"/"w".

Source code in src/pathlib_next/protocols/io.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def open(
    self,
    mode="r",
    buffering=-1,
    encoding: str = None,
    errors: str = None,
    newline: str = None,
) -> _io.IOBase:
    """
    Open the a handle to an object that implement io.IOBase

    The mode is validated like `open()` (ValueError for an invalid mode,
    or an encoding/errors/newline with a binary one), and `_open()`
    receives it canonical: one of "r"/"w"/"x"/"a", plus "+" if given --
    never "b" or "t", so "rt"/"wt" reach every backend as "r"/"w".
    """
    mode = _canonical_mode(mode, buffering, encoding, errors, newline)
    fh = self._open(mode.replace("b", ""), buffering)
    if "b" not in mode:
        try:
            # io.text_encoding is 3.10+; on 3.9 pass encoding through
            # as-is (None means locale default, same effective behavior).
            encoding = getattr(_io, "text_encoding", lambda e, stacklevel=1: e)(
                encoding
            )
            # buffering=1 is line buffering in text mode, as for open().
            fh = _io.TextIOWrapper(
                fh, encoding, errors, newline, line_buffering=buffering == 1
            )
        except BaseException:
            # An unknown encoding (LookupError) or newline (ValueError)
            # fails after the backend handle is open; io.open closes its
            # raw file on that path, and so must we -- a leaked write
            # handle could otherwise commit an empty upload at GC time.
            fh.close()
            raise
    return fh

read_bytes()

Open in bytes mode, read it, and close the file.

Source code in src/pathlib_next/protocols/io.py 
101
102
103
104
105
106
def read_bytes(self) -> bytes:
    """
    Open in bytes mode, read it, and close the file.
    """
    with self.open(mode="rb") as f:
        return f.read()

read_text(encoding=None, errors=None, newline=None)

Open in text mode, read it, and close the file. (newline= is 3.13 parity.)

Source code in src/pathlib_next/protocols/io.py 
108
109
110
111
112
113
114
115
116
117
118
def read_text(
    self, encoding: str = None, errors: str = None, newline: str = None
) -> str:
    """
    Open in text mode, read it, and close the file. (newline= is 3.13
    parity.)
    """
    with self.open(
        mode="r", encoding=encoding, errors=errors, newline=newline
    ) as f:
        return f.read()

write_bytes(data)

Open in bytes mode, write to it, and close the file.

Source code in src/pathlib_next/protocols/io.py 
120
121
122
123
124
125
126
127
def write_bytes(self, data: bytes):
    """
    Open in bytes mode, write to it, and close the file.
    """
    # type-check for the buffer interface before truncating the file
    view = memoryview(data)
    with self.open(mode="wb") as f:
        return f.write(view)

write_text(data, encoding=None, errors=None, newline=None)

Open in text mode, write to it, and close the file.

Source code in src/pathlib_next/protocols/io.py 
129
130
131
132
133
134
135
136
137
138
139
140
def write_text(
    self, data: str, encoding: str = None, errors: str = None, newline: str = None
):
    """
    Open in text mode, write to it, and close the file.
    """
    if not isinstance(data, str):
        raise TypeError("data must be str, not %s" % data.__class__.__name__)
    with self.open(
        mode="w", encoding=encoding, errors=errors, newline=newline
    ) as f:
        return f.write(data)

pathlib_next.protocols.checksum

NativeChecksum

Bases: Protocol

Composable protocol for backends that can compute a file digest server-side, without streaming the file's content through this process.

Optional per-Path-subclass, like every other protocol in this package (io.BinaryOpen, fs.Stat, fs.Chmod) -- most backends will never implement it, and the base Path ABC does not require it. utils.sync.PathSyncer (and any other caller) is expected to try this first and fall back to a streaming checksum (utils.checksum.md5/ sha256) on NotImplementedError.

Two methods, two different jobs: supported_checksums() is an advisory capability query (never raises, so a caller can pick a shared algorithm across two paths -- e.g. source.supported_checksums() & target.supported_checksums() -- before calling anything expensive); checksum() is the authoritative per-call contract and still MUST raise NotImplementedError on its own for any algorithm it can't actually produce, even if a caller never consulted supported_checksums() first (belt-and-suspenders -- the checksum method's own contract, not just the advertisement, is what a correctness-sensitive caller can rely on).

checksum(algorithm='md5')

Return a hex-digest checksum of this file's content, computed by the backend itself (e.g. an SFTP server implementing the filexfer draft's check-file-handle extension -- OpenSSH does not --, a WebDAV getetag, ...) rather than by streaming the content through open("rb").

algorithm names a hashlib-style digest (at least "md5" must be accepted, matching PathSyncer's current default). A backend that cannot produce the requested algorithm -- whether because it supports no native hashing at all, or only a different algorithm -- MUST raise NotImplementedError rather than silently returning a digest under a different algorithm or a value that isn't a true content hash. This is a hard requirement, not a style preference: comparing two checksums computed under different algorithms (or comparing a real hash to something hash-shaped but not actually a content digest, e.g. S3's ETag for a multipart upload) can silently produce a false "in sync" verdict. Callers must never catch anything broader than NotImplementedError here. supported_checksums() not listing algorithm is advisory, not a substitute for this method enforcing its own contract.

Source code in src/pathlib_next/protocols/checksum.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
@_utils.notimplemented
def checksum(self, algorithm: str = "md5") -> str:
    """Return a hex-digest checksum of this file's content, computed by
    the backend itself (e.g. an SFTP server implementing the filexfer draft's
    `check-file-handle` extension -- OpenSSH does not --, a WebDAV `getetag`, ...) rather than by streaming the
    content through `open("rb")`.

    `algorithm` names a `hashlib`-style digest (at least `"md5"` must
    be accepted, matching `PathSyncer`'s current default). A backend
    that cannot produce the requested algorithm -- whether because it
    supports no native hashing at all, or only a different algorithm --
    MUST raise `NotImplementedError` rather than silently returning a
    digest under a different algorithm or a value that isn't a true
    content hash. This is a hard requirement, not a style preference:
    comparing two checksums computed under different algorithms (or
    comparing a real hash to something hash-shaped but not actually a
    content digest, e.g. S3's ETag for a multipart upload) can silently
    produce a false "in sync" verdict. Callers must never catch
    anything broader than `NotImplementedError` here.
    `supported_checksums()` not listing `algorithm` is advisory, not a
    substitute for this method enforcing its own contract.
    """
    ...

supported_checksums()

Advisory set of algorithm names this path can currently produce a native digest for (e.g. frozenset({"md5"})), so a caller can pick a shared algorithm across two paths -- or fall back to streaming -- without probing via trial-and-NotImplementedError per candidate. Never raises. The default (no override, matching the base Path/Pathname ABC not implementing this protocol at all) is an empty frozenset() -- "no native support" -- which is also the correct answer for a backend whose native-hashing capability can vary at runtime (e.g. per-connection extension negotiation) and currently has none available.

This can be a live/dynamic check (e.g. reflecting whether the current server connection actually advertised a given SFTP extension), not necessarily a hardcoded class-level constant -- implementations should recompute it if capability can change within the object's lifetime, and are free to cache it if not.

Source code in src/pathlib_next/protocols/checksum.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def supported_checksums(self) -> "_ty.FrozenSet[str]":
    """Advisory set of `algorithm` names this path can currently
    produce a native digest for (e.g. `frozenset({"md5"})`), so a
    caller can pick a shared algorithm across two paths -- or fall back
    to streaming -- without probing via trial-and-`NotImplementedError`
    per candidate. Never raises. The default (no override, matching the
    base `Path`/`Pathname` ABC not implementing this protocol at all)
    is an empty `frozenset()` -- "no native support" -- which is also
    the correct answer for a backend whose native-hashing capability
    can vary at runtime (e.g. per-connection extension negotiation) and
    currently has none available.

    This can be a live/dynamic check (e.g. reflecting whether the
    current server connection actually advertised a given SFTP
    extension), not necessarily a hardcoded class-level constant --
    implementations should recompute it if capability can change
    within the object's lifetime, and are free to cache it if not.
    """
    return frozenset()