Skip to content

Protocols API

The small protocols Path is built from. A class implements the primitives of the protocols it lists (stat, _open, iterdir, unlink, readlink, ...) and inherits everything derived from them; supports() and capabilities() say which primitives a class has.

pathapi.protocols

supports(obj, *protocols)

Whether obj (an instance or a class) implements every protocol named: each primitive of each resolves to something other than the protocol's default stub. A class whose answer depends on its connection defines _supports(protocol) -> bool | None, asked first for an instance; None falls through to the answer from the class. That hook makes at most one request per backend, cached for the backend's lifetime, and never raises: a failed request answers None.

The answer is advisory, like supported_checksums(): a primitive may still raise NotImplementedError for an argument it cannot handle. TypeError for no protocol, or for something that is not a protocol of this package.

Source code in src/pathapi/protocols/_support.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def supports(obj: object, *protocols: type) -> bool:
    """Whether `obj` (an instance or a class) implements every protocol
    named: each primitive of each resolves to something other than the
    protocol's default stub. A class whose answer depends on its connection
    defines `_supports(protocol) -> bool | None`, asked first for an instance;
    `None` falls through to the answer from the class. That hook makes at most
    one request per backend, cached for the backend's lifetime, and never
    raises: a failed request answers `None`.

    The answer is advisory, like `supported_checksums()`: a primitive may
    still raise `NotImplementedError` for an argument it cannot handle.
    `TypeError` for no protocol, or for something that is not a protocol of
    this package.
    """
    if not protocols:
        raise TypeError("supports() needs at least one protocol")
    for protocol in protocols:
        if not _is_protocol(protocol):
            raise TypeError(f"{protocol!r} is not a protocol of {_PACKAGE}")
    instance = None if isinstance(obj, type) else obj
    cls = obj if isinstance(obj, type) else type(obj)
    for protocol in protocols:
        if instance is not None:
            hook = getattr(cls, "_supports", None)
            if callable(hook):
                answer = hook(instance, protocol)
                if answer is not None:
                    if not answer:
                        return False
                    continue
        if not _static(cls, protocol):
            return False
    return True

capabilities(obj)

The protocols of this package that obj (an instance or a class) supports, in the sense of supports().

Source code in src/pathapi/protocols/_support.py 
113
114
115
116
117
118
119
120
121
122
def capabilities(obj: object) -> "_ty.FrozenSet[type]":
    """The protocols of this package that `obj` (an instance or a class)
    supports, in the sense of `supports()`."""
    module = _importlib.import_module(_PACKAGE + ".protocols")
    known = [
        value
        for value in (getattr(module, name) for name in module.__all__)
        if _is_protocol(value)
    ]
    return frozenset(protocol for protocol in known if supports(obj, protocol))

pathapi.protocols.pure

What the helpers that walk, match and sync need from a pure path.

__all__ = ['Named'] module-attribute

Named

Bases: Protocol

The pure half of a path as glob, walk and sync read it: a name, the segments it is made of, a parent, and the means of building and joining paths of the same kind. Structural only: Pathname and pathlib.PurePath satisfy it without naming it, and nothing subclasses it.

name property

parent property

segments property

__truediv__(key)

Source code in src/pathapi/protocols/pure.py 
29
def __truediv__(self, key: "str | _ty.Self") -> _ty.Self: ...

as_posix()

Source code in src/pathapi/protocols/pure.py 
31
def as_posix(self) -> str: ...

joinpath(*args)

Source code in src/pathapi/protocols/pure.py 
27
def joinpath(self, *args: "str | _ty.Self") -> _ty.Self: ...

with_segments(*segments)

Source code in src/pathapi/protocols/pure.py 
25
def with_segments(self, *segments: str) -> _ty.Self: ...

pathapi.protocols.fs

__all__ = ['Chmod', 'Chown', 'Create', 'FileStatLike', 'Owned', 'Stat', 'Utime'] module-attribute

Chmod

Bases: Chown, Utime, Protocol

Changing the permissions of a path. It also inherits Chown and Utime, which it documented before they were separate protocols, so chown(), _chown() and _utime() stay reachable from it; supports(x, Chmod) asks for chmod() only.

__primitives__ = ('chmod',) class-attribute

__slots__ = () class-attribute instance-attribute

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 providers.parse_mode() for why the base is never left implicit.

Every implementation normalizes its argument through providers.parse_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/pathapi/protocols/fs.py 
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
@_stubs.primitive
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
    `providers.parse_mode()` for why the base is never left implicit.

    Every implementation normalizes its argument through
    `providers.parse_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.
    """
    return _stubs._unimplemented("chmod")

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/pathapi/protocols/fs.py 
324
325
326
327
328
329
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)

Chown

Bases: Protocol

Changing who owns a path: _chown() is the primitive and chown() normalizes its arguments.

__primitives__ = (('_chown', 'chown'),) class-attribute

__slots__ = () class-attribute instance-attribute

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 providers.parse_owner()), so the "unchanged" semantics cannot drift between schemes.

Source code in src/pathapi/protocols/fs.py 
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
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 `providers.parse_owner()`), so the "unchanged" semantics cannot drift
    between schemes.
    """
    uid, gid = _permissions.parse_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)

Create

Bases: Stat, Chmod, Utime, Protocol

Creating a directory or a file: mkdir() and touch() on top of the _mkdir() primitive. touch() also uses open() (BinaryOpen) and unlink() (Remove).

__primitives__ = (('_mkdir', 'mkdir'),) class-attribute

__slots__ = () class-attribute instance-attribute

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

Create a new directory at this given path.

Source code in src/pathapi/protocols/fs.py 
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
def mkdir(self, mode=0o777, parents=False, exist_ok=False):
    """
    Create a new directory at this given path.
    """
    try:
        self._mkdir(mode)
    except FileNotFoundError:
        if not parents or self.parent == self:
            raise
        # Parents may already exist (e.g. a sibling branch created them)
        # -- mirror CPython: exist_ok=True for parents, original
        # exist_ok only for the final retry of self.
        self.parent.mkdir(parents=True, exist_ok=True)
        self.mkdir(mode, parents=False, exist_ok=exist_ok)
    except FileExistsError:
        if not exist_ok or not self.is_dir():
            raise

touch(mode=None, exist_ok=True)

Create this file, if it doesn't exist.

Raises FileExistsError if exist_ok is False and the file already exists (pathlib parity). An existing file is never truncated.

Differences from pathlib.Path.touch, which creates through os.open() (so the process umask applies) and bumps an existing file's mtime:

  • mode is applied with chmod() only when passed explicitly, and then verbatim: a remote server's umask is unknowable. A backend that cannot chmod raises NotImplementedError and the file it created is removed again, so a mode that was asked for is never silently dropped. The default (None) leaves a new file with the permissions the backend gives it, instead of chmod'ing it to a world-writable 0o666.
  • An existing file's access and modification times are set to the current time through _utime(); a backend that cannot set a time leaves the file as it is, since rewriting the content to bump it is not a touch. LocalPath uses pathlib's own touch.
Source code in src/pathapi/protocols/fs.py 
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
def touch(self, mode=None, exist_ok=True):
    """
    Create this file, if it doesn't exist.

    Raises FileExistsError if exist_ok is False and the file already
    exists (pathlib parity). An existing file is never truncated.

    Differences from `pathlib.Path.touch`, which creates through
    `os.open()` (so the process umask applies) and bumps an existing
    file's mtime:

    - `mode` is applied with `chmod()` only when passed explicitly, and
      then verbatim: a remote server's umask is unknowable. A backend
      that cannot chmod raises `NotImplementedError` and the file it
      created is removed again, so a mode that was asked for is never
      silently dropped. The default (`None`) leaves a new file with the
      permissions the backend gives it, instead of chmod'ing it to a
      world-writable 0o666.
    - An existing file's access and modification times are set to the
      current time through `_utime()`; a backend that cannot set a time
      leaves the file as it is, since rewriting the content to bump it
      is not a touch. `LocalPath` uses pathlib's own touch.
    """
    # stat() directly, not exists(): exists() reads *any* OSError (a
    # timeout, a dropped connection) as "missing", which let a transient
    # failure fall through to a truncating open("w").
    try:
        self.stat()
    except FileNotFoundError:
        pass
    else:
        if not exist_ok:
            raise _os_error(FileExistsError, _errno.EEXIST, self)
        now = _time.time()
        try:
            self._utime(atime=now, mtime=now)
        except NotImplementedError:
            pass
        return
    try:
        with self.open("x"):
            ...
    except FileExistsError:
        # Created since the stat() above: an existing file, not ours to
        # truncate.
        if not exist_ok:
            raise
        return
    except NotImplementedError:
        # _open() doesn't support "x" (optional per the mode contract),
        # so the stat() above is the only guard before the truncating
        # "w" -- a small TOCTOU window.
        with self.open("w"):
            ...
    if mode is not None:
        try:
            self.chmod(mode)
        except NotImplementedError:
            # The caller asked for a mode this backend cannot apply: leave
            # no file that has other permissions than the ones asked for.
            try:
                self.unlink()
            except (OSError, NotImplementedError):
                pass
            raise

FileStatLike

Bases: Protocol

Minimum properties stat like object should provide

__slots__ = () class-attribute instance-attribute

st_mode abstractmethod property

st_mtime abstractmethod property

st_size abstractmethod property

Owned

Bases: Stat, Protocol

Reading who owns a path. owner() and group() map the st_uid and st_gid of stat() through the primitives _user_name() and _group_name(); a class that has owner() and group() from somewhere else (pathlib) needs neither.

__primitives__ = (('_user_name', 'owner'), ('_group_name', 'group')) class-attribute

__slots__ = () class-attribute instance-attribute

group(*, follow_symlinks=True)

The name of the group that owns the path. NotImplementedError when stat() carries no st_gid or the class does not map group ids to names (_group_name()).

Source code in src/pathapi/protocols/fs.py 
198
199
200
201
202
203
204
205
def group(self, *, follow_symlinks: bool = True) -> str:
    """The name of the group that owns the path. `NotImplementedError`
    when `stat()` carries no `st_gid` or the class does not map group ids
    to names (`_group_name()`)."""
    gid = getattr(self.stat(follow_symlinks=follow_symlinks), "st_gid", None)
    if gid is None:
        raise NotImplementedError("group() requires stat() to provide st_gid")
    return self._group_name(gid)

owner(*, follow_symlinks=True)

The name of the user that owns the path. NotImplementedError when stat() carries no st_uid or the class does not map user ids to names (_user_name()).

Source code in src/pathapi/protocols/fs.py 
189
190
191
192
193
194
195
196
def owner(self, *, follow_symlinks: bool = True) -> str:
    """The name of the user that owns the path. `NotImplementedError`
    when `stat()` carries no `st_uid` or the class does not map user ids
    to names (`_user_name()`)."""
    uid = getattr(self.stat(follow_symlinks=follow_symlinks), "st_uid", None)
    if uid is None:
        raise NotImplementedError("owner() requires stat() to provide st_uid")
    return self._user_name(uid)

Stat

Bases: Protocol

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

__primitives__ = ('stat',) class-attribute

__slots__ = () class-attribute instance-attribute

info property

What this path is, as an object shaped like pathlib.types.PathInfo (Python 3.14): exists(), is_dir(), is_file() (each with follow_symlinks=True) and is_symlink(). It asks stat() at most once with and once without following links and remembers the answers on that object only: a new info is a new question. LocalPath has pathlib's own on 3.14.

exists(*, follow_symlinks=True)

Whether this path exists.

Source code in src/pathapi/protocols/fs.py 
68
69
70
71
72
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/pathapi/protocols/fs.py 
94
95
96
97
98
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/pathapi/protocols/fs.py 
100
101
102
103
104
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/pathapi/protocols/fs.py 
74
75
76
77
78
79
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_dir_binding()

Whether this directory is another tree's second NAME rather than part of this one: a Windows junction or a mount point (bind or otherwise).

The distinction that matters to anything walking a tree. A symlink announces itself -- is_symlink() is True and every walker here already declines to follow one. A binding does not: it looks like an ordinary directory, and its contents belong to whoever mounted or junctioned it. rm(recursive=True) removes the binding itself rather than the contents behind it.

An implementation that cannot answer (stdlib's is_mount() raises on Windows before 3.12) counts as "not a binding" rather than breaking the operation that asked.

Source code in src/pathapi/protocols/fs.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
def is_dir_binding(self) -> bool:
    """Whether this directory is another tree's second NAME rather than
    part of this one: a Windows junction or a mount point (bind or
    otherwise).

    The distinction that matters to anything walking a tree. A symlink
    announces itself -- `is_symlink()` is True and every walker here
    already declines to follow one. A binding does not: it looks like an
    ordinary directory, and its contents belong to whoever mounted or
    junctioned it. `rm(recursive=True)` removes the binding itself
    rather than the contents behind it.

    An implementation that cannot answer (stdlib's `is_mount()` raises
    on Windows before 3.12) counts as "not a binding" rather than
    breaking the operation that asked.
    """
    for test in (self.is_junction, self.is_mount):
        try:
            if test():
                return True
        except (NotImplementedError, OSError, ValueError):
            continue
    return False

is_fifo()

Whether this path is a FIFO.

Source code in src/pathapi/protocols/fs.py 
106
107
108
109
110
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/pathapi/protocols/fs.py 
81
82
83
84
85
86
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_junction()

Whether this path is a Windows junction (pathlib 3.12 parity).

A junction is NOT a symlink: it is a second NAME for a directory, the Windows equivalent of a Linux bind mount. is_symlink() is False for one, readlink() does not describe it, and a non-following stat reports a plain directory -- which is exactly why a symlink check cannot protect a recursive walk from it.

Default False; LocalPath answers for real. A remote backend has no such concept unless it says otherwise.

Source code in src/pathapi/protocols/fs.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
def is_junction(self) -> bool:
    """Whether this path is a Windows junction (pathlib 3.12 parity).

    A junction is NOT a symlink: it is a second NAME for a directory,
    the Windows equivalent of a Linux bind mount. `is_symlink()` is
    False for one, `readlink()` does not describe it, and a
    non-following stat reports a plain directory -- which is exactly why
    a symlink check cannot protect a recursive walk from it.

    Default False; `LocalPath` answers for real. A remote
    backend has no such concept unless it says otherwise.
    """
    return False

is_mount()

Whether this path is a mount point (pathlib parity).

The POSIX half of the same idea: a bind mount makes one tree visible under a second name, with no link anywhere to say so. Default False; LocalPath answers for real.

Source code in src/pathapi/protocols/fs.py 
142
143
144
145
146
147
148
149
def is_mount(self) -> bool:
    """Whether this path is a mount point (pathlib parity).

    The POSIX half of the same idea: a bind mount makes one tree visible
    under a second name, with no link anywhere to say so. Default False;
    `LocalPath` answers for real.
    """
    return False

is_socket()

Whether this path is a socket.

Source code in src/pathapi/protocols/fs.py 
112
113
114
115
116
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/pathapi/protocols/fs.py 
88
89
90
91
92
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/pathapi/protocols/fs.py 
44
45
46
47
48
49
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)

stat(*, follow_symlinks=True)

Return the stat of the path (a stub: NotImplementedError).

Source code in src/pathapi/protocols/fs.py 
39
40
41
42
@_stubs.primitive
def stat(self, *, follow_symlinks=True) -> FileStatLike:
    """Return the stat of the path (a stub: `NotImplementedError`)."""
    return _stubs._unimplemented("stat")

Utime

Bases: Protocol

Setting the access and modification times of a path.

__primitives__ = ('_utime',) class-attribute

__slots__ = () class-attribute instance-attribute

pathapi.protocols.io

__all__ = ['BinaryOpen'] module-attribute

BinaryOpen

Bases: Protocol

Protocol for objects that support open->io.IoBase

__primitives__ = (('_open', 'open'),) class-attribute

__slots__ = () class-attribute instance-attribute

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/pathapi/protocols/io.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
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".

A stream that commits on close inherits providers.DiscardOnError, so that its own __exit__ calls discard() before closing it when the with block ended in an exception; binary mode returns the stream _open() gave it as it is. In text mode the wrapper open() makes does the same for any stream that has a callable discard().

Source code in src/pathapi/protocols/io.py 
 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
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
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".

    A stream that commits on close inherits `providers.DiscardOnError`, so
    that its own `__exit__` calls `discard()` before closing it when the
    `with` block ended in an exception; binary mode returns the stream
    `_open()` gave it as it is. In text mode the wrapper `open()` makes
    does the same for any stream that has a callable `discard()`.
    """
    mode = _canonical_mode(mode, buffering, encoding, errors, newline)
    fh = self._open(mode.replace("b", ""), buffering)
    if "b" in mode:
        return fh
    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().
        discard = getattr(fh, "discard", None)
        if callable(discard):
            return _discard.DiscardingTextIO(
                fh,
                discard,
                encoding,
                errors,
                newline,
                line_buffering=buffering == 1,
            )
        return _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 then, and so must we: a leaked write
        # handle could commit an empty upload at GC time. A stream that commits on close
        # discards first, so the failed call does not create or replace the file.
        _discard.abandon(fh)
        raise

read_bytes()

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

Source code in src/pathapi/protocols/io.py 
124
125
126
127
128
129
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/pathapi/protocols/io.py 
131
132
133
134
135
136
137
138
139
140
141
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/pathapi/protocols/io.py 
143
144
145
146
147
148
149
150
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/pathapi/protocols/io.py 
152
153
154
155
156
157
158
159
160
161
162
163
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)

pathapi.protocols.tree

Listing a directory and removing what is in it.

__all__ = ['Listing', 'Remove'] module-attribute

Listing

Bases: Stat, Protocol

Listing the entries of a directory: iterdir() is the primitive; _scandir(), glob(), rglob(), walk() and iteration derive from it.

__primitives__ = ('iterdir',) class-attribute

__slots__ = () class-attribute instance-attribute

__iter__()

Source code in src/pathapi/protocols/tree.py 
29
30
def __iter__(self):
    return self.iterdir()

glob(pattern, *, case_sensitive=None, include_hidden=True, recursive=None, dironly=None, recurse_symlinks=False, native=True, on_error=None, bound_loops=False)

Iterate over this subtree and yield all existing files (of any kind, including directories) matching the given relative pattern.

Pathlib semantics: hidden entries are matched (include_hidden=False filters them out and skips hidden directories), a trailing separator selects directories only, "" does not descend into a directory symlink by default, and a trailing "" also selects files on Python 3.13+ (directories only before). A missing or non-directory parent selects nothing. An empty pattern raises ValueError, an absolute one NonRelativePatternError (a NotImplementedError and a ValueError).

recurse_symlinks=True lets "**" descend through symlinks to directories, as pathlib 3.13 does. Where that would loop it stops: a link to a directory the descent is already inside (compared by st_dev/st_ino) is an entry but is not entered, and where the stat carries no identity no more than 40 links are followed in one descent. A backend without symlinks is not affected.

pattern=None expands the pattern THIS PATH CARRIES (LocalPath("/etc/*.conf").glob(None)) instead of applying one to a directory: the path is split at its first wildcard and globbed from there. "" still raises, as pathlib does. Every keyword means what it does for a pattern argument -- recursive=None auto-enables recursion for a carried "**", and native= applies. include_hidden keeps THIS method's default (True); the module-level glob.glob() this delegates to defaults it to False, stdlib-glob-style, so the two spellings differ on hidden names unless the keyword is passed.

native=True (the default) follows the running interpreter on the two rules pathlib changed mid-series: a trailing "/" is ignored before 3.11, and a component that merely contains "" ("a") raises ValueError before 3.13. native=False applies one rule on every version -- trailing "/" always selects directories only, "a**" is always a plain wildcard -- so a pattern answers the same everywhere, which is what a cross-backend or cross-version caller usually wants.

on_error(error) is called when a directory cannot be listed, the same contract as walk() and os.walk: raising from it propagates (an unreadable directory becomes an error), returning treats that directory as empty. Without it such a listing is skipped silently -- pathlib's behaviour, and the reason a config loader could not tell a missing layer from an unreadable one. error.filename names the directory even when the backend left it empty.

bound_loops=True skips a directory whose (st_dev, st_ino) is already on the current descent path, which is what a loop is: that bounds a Windows junction loop, which no symlink check can see (a junction reports is_symlink() == False, so recurse_symlinks=False does not help and pathlib itself loops until the recursion limit). A directory reachable under two sibling names is entered under both. A backend whose stat carries no identity is walked unbounded.

A "" component auto-enables recursion. Pass recursive=False explicitly to treat "" as a plain "*" instead. Note for remote schemes (http/sftp): a recursive glob walks the whole remote subtree, one request/roundtrip per directory.

Source code in src/pathapi/protocols/tree.py 
 60
 61
 62
 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
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def glob(
    self,
    pattern: str | _ty.Self | None,
    *,
    case_sensitive: bool = None,
    include_hidden: bool = True,
    recursive: bool = None,
    dironly: bool = None,
    recurse_symlinks: bool = False,
    native: bool = True,
    on_error: "_ty.Callable[[OSError], None]" = None,
    bound_loops: bool = False,
):
    """Iterate over this subtree and yield all existing files (of any
    kind, including directories) matching the given relative pattern.

    Pathlib semantics: hidden entries are matched (`include_hidden=False`
    filters them out and skips hidden directories), a trailing separator
    selects directories only, "**" does not descend into a directory
    symlink by default, and a trailing "**" also selects files on Python 3.13+
    (directories only before). A missing or non-directory parent selects
    nothing. An empty pattern raises `ValueError`, an absolute one
    `NonRelativePatternError` (a `NotImplementedError` and a
    `ValueError`).

    `recurse_symlinks=True` lets "**" descend through symlinks to
    directories, as pathlib 3.13 does. Where that would loop it stops: a
    link to a directory the descent is already inside (compared by
    `st_dev`/`st_ino`) is an entry but is not entered, and where the stat
    carries no identity no more than 40 links are followed in one
    descent. A backend without symlinks is not affected.

    `pattern=None` expands the pattern THIS PATH CARRIES
    (`LocalPath("/etc/*.conf").glob(None)`) instead of applying one to a
    directory: the path is split at its first wildcard and globbed from
    there. `""` still raises, as pathlib does. Every keyword means what
    it does for a pattern argument -- `recursive=None` auto-enables
    recursion for a carried "**", and `native=` applies. `include_hidden`
    keeps THIS method's default (True); the module-level
    `glob.glob()` this delegates to defaults it to False,
    stdlib-`glob`-style, so the two spellings differ on hidden names
    unless the keyword is passed.

    `native=True` (the default) follows the running interpreter on the
    two rules pathlib changed mid-series: a trailing "/" is ignored
    before 3.11, and a component that merely contains "**" ("a**")
    raises `ValueError` before 3.13. `native=False` applies one rule on
    every version -- trailing "/" always selects directories only, "a**"
    is always a plain wildcard -- so a pattern answers the same
    everywhere, which is what a cross-backend or cross-version caller
    usually wants.

    `on_error(error)` is called when a directory cannot be listed, the
    same contract as `walk()` and `os.walk`: raising from it propagates
    (an unreadable directory becomes an error), returning treats that
    directory as empty. Without it such a listing is skipped silently --
    pathlib's behaviour, and the reason a config loader could not tell a
    missing layer from an unreadable one. `error.filename` names the
    directory even when the backend left it empty.

    `bound_loops=True` skips a directory whose `(st_dev, st_ino)` is
    already on the current descent path, which is what a loop is: that
    bounds a Windows junction loop, which no symlink check can see (a
    junction reports `is_symlink() == False`, so `recurse_symlinks=False`
    does not help and pathlib itself loops until the recursion limit). A
    directory reachable under two sibling names is entered under both. A
    backend whose stat carries no identity is walked unbounded.

    A "**" component auto-enables recursion. Pass `recursive=False`
    explicitly to treat "**" as a plain "*" instead.
    Note for remote schemes (http/sftp): a recursive glob walks the
    whole remote subtree, one request/roundtrip per directory.
    """
    if pattern is None:
        # The pattern is the path itself; `glob()` splits at the first
        # wildcard, so an absolute one is fine here (unlike a pattern
        # argument, which must stay relative to self).
        return _glob.glob(
            self,
            recursive=recursive,
            include_hidden=include_hidden,
            case_sensitive=case_sensitive,
            dironly=bool(dironly),
            native=native,
            on_error=on_error,
            bound_loops=bound_loops,
            recurse_symlinks=recurse_symlinks,
        )
    # Validates eagerly (like pathlib 3.13+); the returned selection is
    # lazy. The pattern is never joined onto self: `self / pattern` let an
    # absolute pattern escape self and re-parsed "?" as a URI query.
    parts, trailing_sep = _glob.parse_pattern(pattern, native=native)
    if recursive is None:
        recursive = _glob.RECURSIVE in parts
    return _glob.select(
        self,
        parts,
        dironly=trailing_sep if dironly is None else dironly,
        recursive=recursive,
        include_hidden=include_hidden,
        case_sensitive=case_sensitive,
        native=native,
        on_error=on_error,
        bound_loops=bound_loops,
        recurse_symlinks=recurse_symlinks,
    )

iterdir()

Yield path objects of the directory contents.

The children are yielded in arbitrary order, and the special entries '.' and '..' are not included.

Source code in src/pathapi/protocols/tree.py 
32
33
34
35
36
37
38
39
@_stubs.primitive
def iterdir(self) -> "_ty.Iterator[_ty.Self]":
    """Yield path objects of the directory contents.

    The children are yielded in arbitrary order, and the
    special entries '.' and '..' are not included.
    """
    return _stubs._unimplemented("iterdir")

rglob(pattern, *, case_sensitive=None, include_hidden=True, recursive=True, dironly=None, recurse_symlinks=False, native=True, on_error=None, bound_loops=False)

Equivalent to glob(f"**/{pattern}", recursive=True).

pattern=None and native= mean what they do on glob(); with None this is glob(None, recursive=True), since a carried pattern brings its own anchor and nothing can be prefixed to it.

Source code in src/pathapi/protocols/tree.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
def rglob(
    self,
    pattern: str | None,
    *,
    case_sensitive: bool = None,
    include_hidden: bool = True,
    recursive: bool = True,
    dironly: bool = None,
    recurse_symlinks: bool = False,
    native: bool = True,
    on_error: "_ty.Callable[[OSError], None]" = None,
    bound_loops: bool = False,
):
    """Equivalent to `glob(f"**/{pattern}", recursive=True)`.

    `pattern=None` and `native=` mean what they do on `glob()`; with
    `None` this is `glob(None, recursive=True)`, since a carried pattern
    brings its own anchor and nothing can be prefixed to it."""
    if pattern is None:
        return self.glob(
            None,
            case_sensitive=case_sensitive,
            include_hidden=include_hidden,
            recursive=recursive,
            dironly=dironly,
            recurse_symlinks=recurse_symlinks,
            native=native,
            on_error=on_error,
            bound_loops=bound_loops,
        )
    if not (isinstance(pattern, str) and not pattern):
        # Reject an absolute pattern before "**/" hides its anchor; glob() validates
        # without listing. A pattern with no component ("." or "./") is allowed.
        try:
            self.glob(pattern, recursive=False, native=native)
        except NonRelativePatternError:
            raise
        except ValueError:
            pass
    return self.glob(
        f"**/{pattern}",
        case_sensitive=case_sensitive,
        include_hidden=include_hidden,
        recursive=recursive,
        dironly=dironly,
        recurse_symlinks=recurse_symlinks,
        native=native,
        on_error=on_error,
        bound_loops=bound_loops,
    )

walk(top_down=True, on_error=None, follow_symlinks=False)

Walk the directory tree from this directory, similar to os.walk().

Uses _scandir() rather than iterdir() + a stat() per entry -- for schemes whose listing already carries type metadata (HTTP/DAV indexes, SFTP listdir_attr, FTP MLSD, S3 list pages), this turns a remote-tree walk from O(entries) round trips into O(dirs). The pre-seeded stat is trusted only when follow_symlinks is False (matching walk()'s own default and the lstat-like semantics of those listing calls); an explicit follow_symlinks=True always re-stat()s each entry so a symlink is still resolved.

Without follow_symlinks a symlink to a directory is listed in filenames and not entered, and so is a Windows junction (is_junction()): like a symlink it is a second NAME for another tree, which pathlib.Path.walk() does not enter either (a junction that points at its own parent would otherwise be walked until the path is too long). Nothing protects a walk with follow_symlinks=True from a loop.

A listed name that is not one path component ("", ., .., or one containing / or NUL; see providers.is_safe_child_name()) is left out of dirnames and filenames. on_error, when given, is called with a ValueError for it, error.filename naming the directory.

Source code in src/pathapi/protocols/tree.py 
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
def walk(
    self,
    top_down=True,
    on_error: _ty.Callable[[OSError | ValueError], None] = None,
    follow_symlinks=False,
):
    """Walk the directory tree from this directory, similar to os.walk().

    Uses `_scandir()` rather than `iterdir()` + a `stat()` per entry --
    for schemes whose listing already carries type metadata (HTTP/DAV
    indexes, SFTP `listdir_attr`, FTP MLSD, S3 list pages), this turns a
    remote-tree walk from O(entries) round trips into O(dirs). The
    pre-seeded stat is trusted only when `follow_symlinks` is False
    (matching `walk()`'s own default and the lstat-like semantics of
    those listing calls); an explicit `follow_symlinks=True` always
    re-`stat()`s each entry so a symlink is still resolved.

    Without `follow_symlinks` a symlink to a directory is listed in
    `filenames` and not entered, and so is a Windows junction
    (`is_junction()`): like a symlink it is a second NAME for another
    tree, which `pathlib.Path.walk()` does not enter either (a junction
    that points at its own parent would otherwise be walked until the
    path is too long). Nothing protects a walk with `follow_symlinks=True`
    from a loop.

    A listed name that is not one path component (`""`, `.`, `..`, or one
    containing `/` or NUL; see `providers.is_safe_child_name()`) is left out
    of `dirnames` and `filenames`. `on_error`, when given, is called with
    a `ValueError` for it, `error.filename` naming the directory.
    """
    paths: "list[_ty.Self|tuple[_ty.Self, list[str], list[str]]]" = [self]

    while paths:
        path = paths.pop()
        if isinstance(path, tuple):
            yield path
            continue
        try:
            # Materialized inside the try: `_scandir()` is a generator, so a failing
            # listing may raise at the first `next()`, and `on_error` must see it.
            entries = list(path._scandir())
        except OSError as error:
            if on_error is not None:
                on_error(error)
            continue

        dirnames: "list[str]" = []
        filenames: "list[str]" = []
        for name, stat in entries:
            if not _names.is_safe_child_name(name):
                # A listing is untrusted input: "..", "a/b" or "" would
                # build a child outside `path` and send the walk there.
                if on_error is not None:
                    error = ValueError(
                        f"refusing unsafe child name {name!r} under {path}"
                    )
                    error.filename = str(path)
                    on_error(error)
                continue
            try:
                if stat is None or follow_symlinks:
                    stat = FileStat.from_path(
                        _entries._child(path, name), follow_symlink=follow_symlinks
                    )
                is_dir = stat.is_dir() if stat is not None else False
                if is_dir and not follow_symlinks:
                    is_dir = not _is_junction(_entries._child(path, name))
            except OSError:
                # Carried over from os.path.isdir().
                is_dir = False

            if is_dir:
                dirnames.append(name)
            else:
                filenames.append(name)

        if top_down:
            yield path, dirnames, filenames
        else:
            paths.append((path, dirnames, filenames))

        paths += [_entries._child(path, d) for d in reversed(dirnames)]

Remove

Bases: Listing, Protocol

Removing a file, an empty directory or a whole tree: unlink() and rmdir() are the primitives; rm() derives from them and from _scandir().

__primitives__ = ('unlink', 'rmdir') class-attribute

__slots__ = () class-attribute instance-attribute

rm(recursive=False, missing_ok=False, ignore_error=False, *, follow_symlinks=False, follow_binds=False)

Remove this file or directory, optionally recursively and ignoring errors.

follow_symlinks applies to a symlink met during a recursive removal, follow_binds to a binding -- a directory that is another tree's second NAME rather than part of this one (is_dir_binding(): a Windows junction, a mount point, a bind mount). Both take:

  • False (default) -- remove the entry itself, never what is behind it: unlink() for a symlink, rmdir() for a binding. This is what rm -r does, and on a live mount the rmdir() fails rather than emptying someone else's filesystem.
  • True -- remove the contents behind it, then the entry. What a walker that cannot tell the difference does by accident.
  • None -- leave it in place. The enclosing directory is then not empty, so its own removal reports that.
  • a callable policy(path) -> bool | None -- asked per entry with the same three answers, so one tree can keep one mount and follow another.

Path components BEFORE the final one are followed as usual; these decide what happens to the entry itself, so they apply to the path rm() is called on as to one met on the way down: a symlink named directly is removed, emptied first (True) or left (None), and without recursive it is only ever unlinked. The name matches stat()/walk()/copy()'s follow_symlinks= rather than inventing a second vocabulary for the same idea.

A policy that is not True, False, None or a callable raises ValueError before anything is removed; so does a callable's answer that is not one of the three, whatever ignore_error says. A listed name that is not one path component ("", ., .., or one containing / or NUL; see providers.is_safe_child_name()) is never joined onto the directory: it raises ValueError, offered to ignore_error as (error, directory) once per entry, and nothing is removed for it.

Source code in src/pathapi/protocols/tree.py 
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
def rm(
    self,
    /,
    recursive=False,
    missing_ok=False,
    ignore_error: bool | _ty.Callable[[Exception, _ty.Self], bool] = False,
    *,
    follow_symlinks: "bool | None | _ty.Callable[[_ty.Self], bool | None]" = False,
    follow_binds: "bool | None | _ty.Callable[[_ty.Self], bool | None]" = False,
):
    """Remove this file or directory, optionally recursively and ignoring errors.

    `follow_symlinks` applies to a symlink met during a recursive
    removal, `follow_binds` to a binding -- a directory that is another
    tree's second NAME rather than part of this one (`is_dir_binding()`:
    a Windows junction, a mount point, a bind mount). Both take:

    - `False` (default) -- remove the entry itself, never what is behind
      it: `unlink()` for a symlink, `rmdir()` for a binding. This is what
      `rm -r` does, and on a live mount the `rmdir()` fails rather than
      emptying someone else's filesystem.
    - `True` -- remove the contents behind it, then the entry. What a
      walker that cannot tell the difference does by accident.
    - `None` -- leave it in place. The enclosing directory is then not
      empty, so its own removal reports that.
    - a callable `policy(path) -> bool | None` -- asked per entry with
      the same three answers, so one tree can keep one mount and follow
      another.

    Path components BEFORE the final one are followed as usual; these
    decide what happens to the entry itself, so they apply to the path
    `rm()` is called on as to one met on the way down: a symlink named
    directly is removed, emptied first (`True`) or left (`None`), and
    without `recursive` it is only ever unlinked. The name matches
    `stat()`/`walk()`/`copy()`'s `follow_symlinks=` rather than
    inventing a second vocabulary for the same idea.

    A policy that is not `True`, `False`, `None` or a callable raises
    `ValueError` before anything is removed; so does a callable's answer
    that is not one of the three, whatever `ignore_error` says. A listed
    name that is not one path component (`""`, `.`, `..`, or one
    containing `/` or NUL; see `providers.is_safe_child_name()`) is never
    joined onto the directory: it raises `ValueError`, offered to
    `ignore_error` as `(error, directory)` once per entry, and nothing is
    removed for it.
    """
    # Same bool-or-callable normalization as copy()/PathSyncer, via the
    # shared helper. A supplied callable keeps rm()'s own `(error, path)`
    # arity -- arities differ per call site by design, see the helper.
    _onerror = _errorpolicy.as_error_handler(ignore_error)
    check_follow("follow_symlinks", follow_symlinks)
    check_follow("follow_binds", follow_binds)

    # An error the handler declined, or raised itself, on its way out:
    # each enclosing directory's `except` catches it again, and consulting
    # the handler there reported one failure once per ancestor.
    declined = []

    def _handle(error, path):
        if isinstance(error, _InvalidPolicy):
            # A mistake of the caller, not a removal that failed.
            raise error
        if any(error is seen for seen in declined):
            raise error
        try:
            swallowed = _onerror(error, path)
        except BaseException as raised:
            declined.append(raised)
            raise
        if not swallowed:
            declined.append(error)
            raise error

    def _remove_tree(path, *, link=False):
        try:
            entries = list(_listing(path))
        except Exception as error:
            _handle(error, path)
            return

        for name, child_stat in entries:
            child = None
            try:
                # Checked before any child is built from the name; only the one-component
                # rule, as a listed name is not rewritten by the platform that listed it.
                if not _names.is_safe_child_name(name):
                    raise ValueError(
                        f"refusing unsafe child name {name!r} under {path}"
                    )
                child = _entries._child(path, name)
                if child_stat is None:
                    child_stat = FileStat.from_path(child, follow_symlink=False)
                if child_stat is not None and child_stat.is_symlink():
                    # A link, whatever it points at. The non-following
                    # stat says "link", so this branch is reached before
                    # the directory one.
                    policy = _follow_policy(follow_symlinks, child)
                    if policy == "ignore":
                        continue
                    if policy == "follow" and child.is_dir():
                        _remove_tree(child, link=True)
                    else:
                        child.unlink()
                elif child_stat is not None and child_stat.is_dir():
                    if child.is_dir_binding():
                        # A binding: what is inside belongs to the tree
                        # it names, not to this one.
                        policy = _follow_policy(follow_binds, child)
                        if policy == "ignore":
                            continue
                        if policy == "follow":
                            _remove_tree(child)
                        else:
                            # `rmdir()` on a live mount fails (EBUSY) --
                            # the right answer, and far better than
                            # emptying someone else's filesystem.
                            child.rmdir()
                    else:
                        _remove_tree(child)
                else:
                    child.unlink()
            except Exception as error:
                _handle(error, path if child is None else child)

        try:
            # `rmdir()` on a symlink is ENOTDIR on POSIX: the link is
            # unlinked once what is behind it has been removed.
            path.unlink() if link else path.rmdir()
        except Exception as error:
            _handle(error, path)

    try:
        stat = FileStat.from_path(self, follow_symlink=False)
    except Exception as error:
        _handle(error, self)
        return
    if stat is None:
        if not missing_ok:
            _handle(_os_error(FileNotFoundError, _errno.ENOENT, self), self)
    elif stat.is_symlink():
        policy = _follow_policy(follow_symlinks, self)
        if policy == "ignore":
            return
        try:
            if policy == "follow" and recursive and self.is_dir():
                _remove_tree(self, link=True)
            else:
                self.unlink()
        except Exception as error:
            _handle(error, self)
    elif stat.is_dir():
        binding = self.is_dir_binding()
        policy = _follow_policy(follow_binds, self) if binding else "follow"
        if policy == "ignore":
            return
        if recursive and (not binding or policy == "follow"):
            _remove_tree(self)
        else:
            try:
                self.rmdir()
            except Exception as error:
                _handle(error, self)
    else:
        try:
            self.unlink()
        except Exception as error:
            _handle(error, self)

rmdir()

Remove this directory. The directory must be empty.

Source code in src/pathapi/protocols/tree.py 
319
320
321
322
323
324
@_stubs.primitive
def rmdir(self):
    """
    Remove this directory.  The directory must be empty.
    """
    return _stubs._unimplemented("rmdir")

Remove this file or link. If the path is a directory, use rmdir() instead.

Source code in src/pathapi/protocols/tree.py 
311
312
313
314
315
316
317
@_stubs.primitive
def unlink(self, missing_ok=False):
    """
    Remove this file or link.
    If the path is a directory, use rmdir() instead.
    """
    return _stubs._unimplemented("unlink")

Symbolic links.

__all__ = ['Hardlink', 'Readlink', 'Symlink'] module-attribute

Bases: Stat, Protocol

Making a path a second name for an existing file: _hardlink_to() is the primitive; hardlink_to() turns a str target into a path of this backend.

__primitives__ = (('_hardlink_to', 'hardlink_to'),) class-attribute

__slots__ = () class-attribute instance-attribute

Make this path a hard link to target (pathlib 3.10 parity). A str is a path on this backend.

Source code in src/pathapi/protocols/links.py 
130
131
132
133
134
135
def hardlink_to(self, target: "_ty.Self | str") -> None:
    """Make this path a hard link to `target` (pathlib 3.10 parity). A
    `str` is a path on this backend."""
    if isinstance(target, str):
        target = self.with_segments(target)
    return self._hardlink_to(target)

Bases: Stat, Protocol

Reading a symbolic link: readlink() is the primitive. A read-only backend (an archive) has this and not Symlink.

__primitives__ = ('readlink',) class-attribute

__slots__ = () class-attribute instance-attribute

The path this symbolic link points to, as a path of this class whose str() and as_posix() are the link text, unresolved (a stub: NotImplementedError). OSError(EINVAL) where the path is not a link.

Source code in src/pathapi/protocols/links.py 
22
23
24
25
26
27
28
@_stubs.primitive
def readlink(self) -> _ty.Self:
    """The path this symbolic link points to, as a path of this class whose
    `str()` and `as_posix()` are the link text, unresolved (a stub:
    `NotImplementedError`). `OSError(EINVAL)` where the path is not a
    link."""
    return _stubs._unimplemented("readlink")

Bases: Remove, Protocol

Making a symbolic link: _symlink_to() is the primitive; symlink_to() normalizes its target and honours force=.

__primitives__ = (('_symlink_to', 'symlink_to'),) class-attribute

__slots__ = () class-attribute instance-attribute

Make this path a symlink pointing to target.

Signature-compatible with pathlib.Path.symlink_to(); force is a pathapi extension (see docs/divergences.md).

With force=True, an existing entry at this path is removed first. No filesystem or transport offers an atomic "replace a symlink" operation, so this is an unlink-then-symlink sequence and is therefore not atomic: between the two steps the path does not exist, and a concurrent writer can win the race. It is a convenience, not a locking primitive.

Only a non-directory entry is removed -- an existing directory at the link path is left alone and the underlying FileExistsError (or the backend's equivalent) propagates. Silently deleting a directory tree is never what force= on a symlink call is asking for.

Source code in src/pathapi/protocols/links.py 
 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
100
101
102
103
104
105
106
107
108
109
110
def symlink_to(
    self,
    target: "_ty.Self | str",
    target_is_directory: bool = False,
    *,
    force: bool = False,
) -> None:
    """Make this path a symlink pointing to `target`.

    Signature-compatible with `pathlib.Path.symlink_to()`; `force` is a
    pathapi extension (see `docs/divergences.md`).

    With `force=True`, an existing entry at *this* path is removed
    first. No filesystem or transport offers an atomic "replace a
    symlink" operation, so this is an unlink-then-symlink sequence and
    is therefore **not** atomic: between the two steps the path does not
    exist, and a concurrent writer can win the race. It is a
    convenience, not a locking primitive.

    Only a non-directory entry is removed -- an existing *directory* at
    the link path is left alone and the underlying `FileExistsError`
    (or the backend's equivalent) propagates. Silently deleting a
    directory tree is never what `force=` on a symlink call is asking
    for.
    """
    # Through `_symlink_target()`, not `type(self)(target)`: for a URI-backed path
    # the constructor re-parses the string as URI syntax, cutting the target at "?"
    # or "#" and percent-decoding it.
    target = self._symlink_target(target)
    if force:
        try:
            self.unlink(missing_ok=True)
        except (IsADirectoryError, PermissionError, OSError) as error:
            # A directory at the link path (IsADirectoryError on POSIX; PermissionError or
            # an OSError on Windows and remote backends) is not removed by force=: the
            # symlink attempt below raises the error that describes the conflict.
            if not self.is_dir():
                raise error
    return self._symlink_to(target, target_is_directory)

pathapi.protocols.resolve

Making a path absolute and following the links on its way.

__all__ = ['Resolve'] module-attribute

Resolve

Bases: Readlink, Protocol

absolute() and resolve(). _cwd() is the primitive: the directory a relative path is relative to. resolve() follows links through readlink() where the class has it and is lexical otherwise.

__primitives__ = (('_cwd', 'absolute'),) class-attribute

__slots__ = () class-attribute instance-attribute

absolute()

This path when it is absolute, else _cwd() joined with it.

Source code in src/pathapi/protocols/resolve.py 
35
36
37
38
39
def absolute(self) -> _ty.Self:
    """This path when it is absolute, else `_cwd()` joined with it."""
    if self.is_absolute():
        return self
    return self._cwd().joinpath(self)

resolve(strict=False)

The absolute path with .. applied and every link replaced by what readlink() says, at most 40 of them. strict=True raises FileNotFoundError for a component that is missing; otherwise the rest of the names are kept as they are. A link loop raises what pathlib raises on this interpreter: RuntimeError before 3.13, OSError(ELOOP) from 3.13 when strict (otherwise the looping link is kept).

Source code in src/pathapi/protocols/resolve.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
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
def resolve(self, strict: bool = False) -> _ty.Self:
    """The absolute path with `..` applied and every link replaced by what
    `readlink()` says, at most 40 of them. `strict=True` raises
    `FileNotFoundError` for a component that is missing; otherwise the
    rest of the names are kept as they are. A link loop raises what
    `pathlib` raises on this interpreter: `RuntimeError` before 3.13,
    `OSError(ELOOP)` from 3.13 when `strict` (otherwise the looping link
    is kept)."""
    path = self.absolute()
    links = supports(self, Readlink)
    root = path.with_segments(path.anchor or "/")
    resolved = root
    pending = [name for name in path.segments if name][::-1]
    followed = 0
    while pending:
        name = pending.pop()
        if name == ".":
            continue
        if name == "..":
            resolved = resolved.parent
            continue
        candidate = resolved.joinpath(name)
        if not links:
            if strict:
                candidate.stat()
            resolved = candidate
            continue
        try:
            mode = candidate.stat(follow_symlinks=False).st_mode
        except OSError:
            if strict:
                raise
            resolved = candidate
            continue
        if not _stat.S_ISLNK(mode):
            resolved = candidate
            continue
        followed += 1
        if followed > _MAX_LINKS:
            if _sys.version_info < (3, 13):
                raise RuntimeError(f"Symlink loop from {candidate!r}")
            if strict:
                raise OSError(
                    _errno.ELOOP,
                    "Too many levels of symbolic links",
                    str(candidate),
                )
            resolved = candidate
            continue
        target = candidate.readlink()
        text = getattr(target, "path", None)
        if not isinstance(text, str):
            text = target.as_posix()
        if text.startswith("/"):
            resolved = root
        pending.extend(part for part in reversed(text.split("/")) if part)
    return resolved

pathapi.protocols.transfer

Telling two paths apart, copying and moving.

__all__ = ['Copy', 'Identity', 'Move'] module-attribute

Copy

Bases: Identity, Create, Symlink, Readlink, BinaryOpen, Protocol

Copying a file or a tree to another path, of this backend or of another. It precedes BinaryOpen in a class's bases because both define copy() and this one takes the options of a path copy.

__primitives__ = () class-attribute

__slots__ = () class-attribute instance-attribute

copy(target, *, overwrite=False, follow_symlinks=True, preserve_metadata=True, recursive=False, ignore_error=None, progress=None, on_error=None)

Copy this file's content to target.

follow_symlinks/preserve_metadata are named to match CPython 3.14's Path.copy() (added after this method); overwrite is our own extension (3.14 has no equivalent -- it always raises if the destination exists). preserve_metadata defaults to True here (unlike 3.14's False): st_mode is propagated, but not timestamps or xattrs -- full metadata preservation is not implemented. ignore_error accepts a bool or a callable. True ignores every error; False and None (the default) fail on the first error. A callable is invoked as ignore_error(error) and, as it always has here, is a notification hook: the error is suppressed regardless of what it returns, so handlers like errors.append (returning None) keep working. Only errors from child copies during a recursive=True copy are routed here.

on_error is rm()'s predicate: a callable (error, path) -> bool (or a bool) asked once per child error, with path the source entry that failed. True skips the entry and goes on, False raises the error, and a predicate that raises ends the copy with its own exception. An error it declined is not offered again by the directories above. ignore_error and on_error are two spellings of one policy: passing both raises TypeError before anything is copied.

A child whose listed name is not one safe component for target (a drive such as C:x or a backslash on a Windows target, which land outside it; a trailing dot, which Windows stores under another name) is refused with ValueError and reported through ignore_error or on_error like any child failure. PathSyncer and unpack_archive() apply the same rule; an archive listing keeps such names, as they are ordinary filenames on POSIX.

progress, when given, is called as progress(path, bytes_copied, total_size) for every chunk written during each file copy (path is the source Path being streamed -- self for a single-file copy, or the relevant child during a recursive=True copy). bytes_copied increases monotonically per file and reaches total_size (or None if the size couldn't be determined) at the end of that file. Directories themselves don't get a progress call (only the files inside them do). With progress=None (the default), behavior is unchanged -- no per-chunk overhead. Native backend transfers that bypass the generic streaming copy (e.g. an SFTP backend's concurrent fan-out) do not invoke progress; see docs/divergences.md's "Deliberate extensions" section.

Source code in src/pathapi/protocols/transfer.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
def copy(
    self,
    target: "Path | str",
    *,
    overwrite=False,
    follow_symlinks=True,
    preserve_metadata=True,
    recursive=False,
    ignore_error=None,
    progress: "_ty.Callable[[_ty.Self, int, _ty.Optional[int]], None]" = None,
    on_error: "bool | _ty.Callable[[Exception, _ty.Self], bool] | None" = None,
):
    """Copy this file's content to `target`.

    `follow_symlinks`/`preserve_metadata` are named to match CPython
    3.14's Path.copy() (added after this method); `overwrite` is our
    own extension (3.14 has no equivalent -- it always raises if the
    destination exists). `preserve_metadata` defaults to True here
    (unlike 3.14's False): st_mode is propagated, but not
    timestamps or xattrs -- full metadata preservation is not implemented.
    `ignore_error` accepts a bool or a callable. `True` ignores every
    error; `False` and `None` (the default) fail on the first error. A
    callable is invoked as `ignore_error(error)` and, as it always has
    here, is a *notification* hook: the error is suppressed regardless of
    what it returns, so handlers like `errors.append` (returning None)
    keep working. Only errors from *child* copies during a
    `recursive=True` copy are routed here.

    `on_error` is `rm()`'s predicate: a callable `(error, path) -> bool`
    (or a bool) asked once per child error, with `path` the source entry
    that failed. `True` skips the entry and goes on, `False` raises the
    error, and a predicate that raises ends the copy with its own
    exception. An error it declined is not offered again by the
    directories above. `ignore_error` and `on_error` are two spellings of
    one policy: passing both raises `TypeError` before anything is copied.

    A child whose listed name is not one safe component for `target` (a
    drive such as `C:x` or a backslash on a Windows target, which land
    outside it; a trailing dot, which Windows stores under another name)
    is refused with `ValueError` and reported through `ignore_error` or
    `on_error` like any child failure. `PathSyncer` and
    `unpack_archive()` apply the same rule; an archive listing keeps
    such names, as they are ordinary filenames on POSIX.

    `progress`, when given, is called as `progress(path, bytes_copied,
    total_size)` for every chunk written during each *file* copy (`path`
    is the source `Path` being streamed -- `self` for a single-file
    copy, or the relevant child during a `recursive=True` copy).
    `bytes_copied` increases monotonically per file and reaches
    `total_size` (or `None` if the size couldn't be determined) at the
    end of that file. Directories themselves don't get a progress call
    (only the files inside them do). With `progress=None` (the
    default), behavior is unchanged -- no per-chunk overhead. Native
    backend transfers that bypass the generic streaming copy (e.g.
    an SFTP backend's concurrent fan-out) do not invoke `progress`;
    see `docs/divergences.md`'s "Deliberate extensions" section.
    """
    offering = copy_error_predicate(ignore_error, on_error)
    if isinstance(target, str):
        target = self._coerce_target(target)
    src = self
    handed = _take_handed(src, target)
    known = None if handed is None else handed.source_stat

    if not follow_symlinks and known is None and src.is_symlink():
        # pathlib 3.14: copy the link itself, not what it points at.
        # Copying the target's content (and chmod'ing it with the link's
        # own 0o777) silently defeated the flag.
        return src._copy_symlink(target, overwrite=overwrite)

    # One stat of the source answers its type, its mode and, beside the
    # target's, whether the two are one file; None when it cannot be had.
    src_stat = known if known is not None else _stat_or_none(src, follow_symlinks)
    target_absent = handed is not None and handed.target_absent

    if recursive and src_stat is not None and _stat.S_ISDIR(src_stat.st_mode):
        if contains(src, target):
            # Checked before anything is created: the new directory would
            # be listed and copied into itself without end.
            raise OSError(
                _errno.EINVAL,
                "Cannot copy a directory into itself",
                str(target),
            )
        # The writes below may be committed together by a backend that
        # pays for each one (see `_write_batch`).
        with _write_batch_of(target):
            # Listed before the target exists, as shutil.copytree does.
            children = list(src.iterdir())
            if not target_absent and target.exists():
                if not target.is_dir():
                    raise FileExistsError(target)
                if not overwrite:
                    raise FileExistsError(target)
            else:
                target.mkdir()
            windows_target = _names.is_windows_flavoured(target)
            for child in children:
                try:
                    # A listed name is untrusted: one that is not a single component inside
                    # `target` is refused (see the child-name paragraph of the docstring).
                    if not _names.is_safe_child_name(
                        child.name, windows=windows_target
                    ):
                        raise ValueError(
                            f"refusing unsafe child name {child.name!r} "
                            f"under {target}"
                        )
                    child.copy(
                        target / child.name,
                        overwrite=overwrite,
                        follow_symlinks=follow_symlinks,
                        preserve_metadata=preserve_metadata,
                        recursive=True,
                        ignore_error=ignore_error,
                        progress=progress,
                        # Passed on only when given, so a `copy()` override
                        # without the keyword still works as a child.
                        **({} if offering is None else {"on_error": offering}),
                    )
                except Exception as e:
                    if offering is not None:
                        if not offering(e, child):
                            raise
                    elif callable(ignore_error):
                        # A notify-and-suppress hook: its return value is not
                        # consulted, so `errors.append` works.
                        ignore_error(e)
                    elif not ignore_error:
                        raise
            if preserve_metadata:
                # After the children: a read-only directory could not take
                # them, and its mode is only settled once they are in.
                _copy_mode(src, target, follow_symlinks, src_stat)
            return

    # What needs no I/O first; the stats below settle the rest.
    if _same_file(src, target, identity=False):
        raise OSError(
            _errno.EINVAL, "Source and target are the same file", str(target)
        )
    # stat(), not exists(): exists() reads a transient error (a 503, a
    # timeout) as "missing", and overwrite=False must never be decided by
    # a failure. Only FileNotFoundError means the target is absent.
    try:
        target_stat = None if target_absent else target.stat()
    except FileNotFoundError:
        target_stat = None
    target_exists = target_stat is not None
    if target_stat is not None:
        if _same_identity(src_stat, target_stat):
            raise OSError(
                _errno.EINVAL, "Source and target are the same file", str(target)
            )
        if _stat.S_ISDIR(target_stat.st_mode):
            raise IsADirectoryError(target)
        if not overwrite:
            raise FileExistsError(target)

    # Open the source before the target is touched at all: a missing or
    # unreadable source must leave an existing target intact and must not
    # leave a new empty one behind.
    with src.open("rb") as input:
        if target_exists and supports(target, Remove):
            target.unlink(missing_ok=True)
        created = False
        try:
            with target.open("wb") as output:
                created = True
                BinaryOpen._copy_stream(
                    src,
                    input,
                    output,
                    progress=(
                        None
                        if progress is None
                        else lambda copied, total: progress(src, copied, total)
                    ),
                    total_size=(
                        _ASK_STAT
                        if src_stat is None
                        else (
                            getattr(src_stat, "st_size", None)
                            if getattr(src_stat, "size_known", True)
                            else None
                        )
                    ),
                )
        except BaseException:
            # A half-written target is not a copy of anything; do not
            # leave it looking like one.
            if created and supports(target, Remove):
                try:
                    target.unlink(missing_ok=True)
                except Exception:
                    pass
            raise

    if preserve_metadata:
        _copy_mode(src, target, follow_symlinks, src_stat)

copy_into(target_dir, *, overwrite=False, follow_symlinks=True, preserve_metadata=True, recursive=False, ignore_error=None, progress=None, on_error=None)

Copy this file or directory into the directory target_dir, under its own name, and return the new path (pathlib 3.14 parity).

copy(target_dir / self.name, ...) with the same keywords and the same defaults as copy(): an existing target needs overwrite=True and a directory source needs recursive=True. ValueError for a path with no name.

Source code in src/pathapi/protocols/transfer.py 
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
def copy_into(
    self,
    target_dir: "Path | str",
    *,
    overwrite=False,
    follow_symlinks=True,
    preserve_metadata=True,
    recursive=False,
    ignore_error=None,
    progress: "_ty.Callable[[_ty.Self, int, _ty.Optional[int]], None]" = None,
    on_error: "bool | _ty.Callable[[Exception, _ty.Self], bool] | None" = None,
) -> "Path":
    """Copy this file or directory into the directory `target_dir`, under
    its own name, and return the new path (pathlib 3.14 parity).

    `copy(target_dir / self.name, ...)` with the same keywords and the
    same defaults as `copy()`: an existing target needs `overwrite=True`
    and a directory source needs `recursive=True`. `ValueError` for a
    path with no name."""
    target = self._into(target_dir)
    self.copy(
        target,
        overwrite=overwrite,
        follow_symlinks=follow_symlinks,
        preserve_metadata=preserve_metadata,
        recursive=recursive,
        ignore_error=ignore_error,
        progress=progress,
        **({} if on_error is None else {"on_error": on_error}),
    )
    return target

Identity

Bases: Stat, Protocol

Whether two paths name the same node, and in which namespace. _same_filesystem() and _node_key() are overridden by a class whose instances can front different stores; samefile() compares what stat() reports.

__primitives__ = () class-attribute

__slots__ = () class-attribute instance-attribute

samefile(other_path)

Return whether other_path is the same or not as this file, by comparing (st_dev, st_ino) when the backend's stat() provides them. Raises NotImplementedError otherwise (e.g. our own FileStat doesn't carry st_dev/st_ino) -- LocalPath gets a real implementation from pathlib.Path via MRO instead of this one.

Source code in src/pathapi/protocols/transfer.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def samefile(self, other_path: str | _ty.Self):
    """Return whether other_path is the same or not as this file, by
    comparing (st_dev, st_ino) when the backend's stat() provides them.
    Raises NotImplementedError otherwise (e.g. our own FileStat doesn't
    carry st_dev/st_ino) -- LocalPath gets a real implementation from
    pathlib.Path via MRO instead of this one.
    """
    from ..path import Path  # `path` imports this module

    # with_segments(), never the bare constructor or `_coerce_target()`:
    # a str is a path on this backend. `type(self)(str)` gave a MemPath
    # a fresh, empty backend, and a URI parse drops this path's host.
    other = (
        other_path
        if isinstance(other_path, Path)
        else self.with_segments(other_path)
    )
    st1 = self.stat()
    st2 = other.stat()
    ident1 = (getattr(st1, "st_dev", None), getattr(st1, "st_ino", None))
    ident2 = (getattr(st2, "st_dev", None), getattr(st2, "st_ino", None))
    if None in ident1 or None in ident2:
        raise NotImplementedError(
            "samefile() requires stat() to provide st_dev/st_ino"
        )
    return ident1 == ident2

Move

Bases: Copy, Remove, Protocol

Renaming and moving: rename() is the primitive, and move() falls back to a copy followed by a removal where it is not available.

__primitives__ = ('rename',) class-attribute

__slots__ = () class-attribute instance-attribute

move(target, *, overwrite=False)

Move this file or directory to target, falling back to copy+unlink/rm if rename is unsupported.

Nothing is removed until the move is known to be allowed: a missing source, a file onto a directory, a target that holds the source (OSError ENOTEMPTY) or lies inside it (EINVAL), and a second name of the same file that is a link (EINVAL) all raise first. The same file under another spelling (a case alias, x.move(x)) is renamed in place. Between two hard links of one file, an existing target needs overwrite=True, and the source name is then removed.

Source code in src/pathapi/protocols/transfer.py 
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
def move(self, target: "Path|str", *, overwrite=False):
    """Move this file or directory to target, falling back to copy+unlink/rm if rename is unsupported.

    Nothing is removed until the move is known to be allowed: a missing
    source, a file onto a directory, a target that holds the source
    (`OSError` `ENOTEMPTY`) or lies inside it (`EINVAL`), and a second
    name of the same file that is a link (`EINVAL`) all raise first. The
    same file under another spelling (a case alias, `x.move(x)`) is
    renamed in place. Between two hard links of one file, an existing
    target needs `overwrite=True`, and the source name is then removed.
    """
    if isinstance(target, str):
        target = self._coerce_target(target)
    src = self
    # rename() only makes sense between paths the same backend can see.
    native = src._rename_compatible(target)

    # Everything that can fail cheaply is checked before the target is
    # touched: a missing source, or a file onto a directory, must not
    # delete the target and only then raise.
    held = _Held(src)
    src_stat = held.source
    if src_stat is None:
        raise FileNotFoundError(
            _errno.ENOENT, "No such file or directory", str(src)
        )
    if not supports(src, Remove) and not (native and supports(src, Move)):
        # Copied and then not removed, a move would leave two files.
        raise NotImplementedError(f"move() of {src!r}: it cannot be removed")
    # The same entry under another spelling (a case-only rename on a
    # case-insensitive filesystem, or `x.move(x)`) is renamed in place:
    # removing the "existing" target would delete the source itself.
    relation = _relation(src, target, follow=False, held=held)
    target_stat = None
    if relation != _SAME:
        target_stat = held.target(target)
        if target_stat is not None and not overwrite:
            raise FileExistsError(target)
        if relation == _INNER:
            raise OSError(
                _errno.EINVAL, "Cannot move a directory into itself", str(target)
            )
        if target_stat is not None:
            if relation == _OUTER:
                raise OSError(_errno.ENOTEMPTY, "Directory not empty", str(target))
            if relation == _OTHER:
                return _drop_second_name(src, src_stat, target, target_stat)
            if target_stat.is_dir() and not target_stat.is_symlink():
                if not src_stat.is_dir():
                    raise IsADirectoryError(target)
                target.rm(recursive=True, missing_ok=True)
            elif (
                native
                and type(target) is type(src)
                and getattr(type(src), "replace", Move.replace) is not Move.replace
            ):
                # A class with its own replace() (os.replace() swaps atomically and leaves the
                # target alone on failure). Path.replace() is built on move(): calling it here
                # would never end.
                try:
                    return src.replace(target)
                except OSError as error:
                    if error.errno != _errno.EXDEV:
                        raise
                    native = False
            else:
                target.unlink(missing_ok=True)

    if native:
        try:
            return src.rename(target)
        except NotImplementedError:
            pass
        except OSError as error:
            # Another filesystem or drive: fall back to copy + delete, as
            # shutil.move and 3.14's Path.move do.
            if error.errno != _errno.EXDEV:
                raise

    # A link is recreated, not read through, wherever the target can hold one, so
    # the result does not depend on rename() being available. The copy is handed
    # what is known (nothing at `target`, `src` not a link); a failed rename changed nothing.
    links = _holds_links(src, target)
    handing = _handing(
        src,
        target,
        held.source_raw if not src_stat.is_symlink() else None,
        relation != _SAME and target_stat is None,
    )
    if src_stat.is_symlink() and links:
        with handing:
            src.copy(target, overwrite=overwrite, follow_symlinks=False)
        src.unlink()
    elif src.is_dir() if src_stat.is_symlink() else src_stat.is_dir():
        with handing:
            src.copy(
                target,
                overwrite=overwrite,
                recursive=True,
                follow_symlinks=not links,
            )
        src.rm(recursive=True)
    else:
        with handing:
            src.copy(target, overwrite=overwrite)
        src.unlink()

move_into(target_dir, *, overwrite=False)

Move this file or directory into the directory target_dir, under its own name, and return the new path (pathlib 3.14 parity).

move(target_dir / self.name, overwrite=overwrite): an existing target needs overwrite=True. ValueError for a path with no name.

Source code in src/pathapi/protocols/transfer.py 
575
576
577
578
579
580
581
582
583
def move_into(self, target_dir: "Path | str", *, overwrite=False) -> "Path":
    """Move this file or directory into the directory `target_dir`, under
    its own name, and return the new path (pathlib 3.14 parity).

    `move(target_dir / self.name, overwrite=overwrite)`: an existing
    target needs `overwrite=True`. `ValueError` for a path with no name."""
    target = self._into(target_dir)
    self.move(target, overwrite=overwrite)
    return target

rename(target)

Rename this file or directory to the given target.

Source code in src/pathapi/protocols/transfer.py 
463
464
465
466
@_stubs.primitive
def rename(self, target: "_ty.Self | str"):
    """Rename this file or directory to the given target."""
    return _stubs._unimplemented("rename")

replace(target)

Rename this file or directory to target, replacing what is there, and return target as a path (pathlib parity).

A file replaces a file and a directory replaces an empty directory; a directory or a file standing in the way of the other kind raises as os.replace() does, and a directory that holds anything raises OSError(ENOTEMPTY) rather than being removed. Built on move(), so it works between any two paths move() accepts; LocalPath keeps stdlib's os.replace().

Source code in src/pathapi/protocols/transfer.py 
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
def replace(self, target: "Path | str") -> "Path":
    """Rename this file or directory to `target`, replacing what is
    there, and return `target` as a path (pathlib parity).

    A file replaces a file and a directory replaces an empty directory;
    a directory or a file standing in the way of the other kind raises
    as `os.replace()` does, and a directory that holds anything raises
    `OSError(ENOTEMPTY)` rather than being removed. Built on `move()`, so
    it works between any two paths `move()` accepts; `LocalPath` keeps
    stdlib's `os.replace()`."""
    if isinstance(target, str):
        target = self._coerce_target(target)
    if target.is_dir() and next(iter(target._scandir()), None) is not None:
        raise OSError(_errno.ENOTEMPTY, "Directory not empty", str(target))
    self.move(target, overwrite=True)
    return target

pathapi.protocols.checksum

__all__ = ['NativeChecksum'] module-attribute

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. sync.PathSyncer (and any other caller) is expected to try this first and fall back to a streaming checksum (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).

__primitives__ = ('checksum',) class-attribute

__slots__ = () class-attribute instance-attribute

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/pathapi/protocols/checksum.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@_stubs.primitive
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.
    """
    return _stubs._unimplemented("checksum")

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/pathapi/protocols/checksum.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
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()