Skip to content

Providers API

What a class built on Path calls: the error and follow policies, the name and mode checks, the commit-on-close buffer, the LRU cache and contains.

pathapi.providers

What a class built on Path calls: error and follow policies, name and mode checks, the split of a name into stem and suffix, the commit-on-close buffer, the LRU cache and the contains test.

UNCHANGED = None module-attribute

__all__ = ['CommitOnClose', 'DiscardOnError', 'LRU', 'OfferedOnce', 'TreePath', 'UNCHANGED', 'as_error_handler', 'as_error_predicate', 'check_follow', 'contains', 'copy_error_predicate', 'is_safe_child_name', 'is_windows_flavoured', 'name_stem', 'name_suffix', 'name_suffixes', 'derived_from', 'notimplemented', 'primitive', 'parse_mode', 'parse_owner'] module-attribute

CommitOnClose(commit, initial=None)

Bases: DiscardOnError, BytesIO

A write buffer held in memory whose content is handed to commit when it is closed, once, if it was written to. For a backend with no way to write a file piecewise (an FTP STOR, a rewrite of an archive): the cost is the whole file in memory for the duration of the write.

initial (an open("r+")) is the file's content, with the position at 0, and commit runs only if the buffer was modified; without it, commit runs at the first close even if nothing was written, which creates an empty file. commit(buffer) runs before the buffer is closed, so it can read from it.

discard() drops the buffer so that a later close() commits nothing; leaving a with block through an exception calls it before closing, and so does collecting the stream unclosed, which also warns.

Source code in src/pathapi/_commit.py
74
75
76
77
78
79
80
81
def __init__(
    self,
    commit: "_ty.Callable[[_io.BytesIO], object]",
    initial: "bytes | None" = None,
):
    super().__init__(b"" if initial is None else initial)
    self._commit = commit
    self._dirty = initial is None

close()

Source code in src/pathapi/_commit.py
107
108
109
110
111
112
113
114
115
116
def close(self):
    if self.closed:
        return
    try:
        if self._dirty and not self._discarded:
            self._commit(self)
    finally:
        # Closed even when the commit fails, so `IOBase.__del__` does not
        # retry it (over newer content) at garbage collection.
        super().close()

discard()

Drop what was written: close() commits nothing, whatever is written after this. Calling it twice, or after close(), does nothing more.

Source code in src/pathapi/_commit.py
 98
 99
100
101
102
103
104
105
def discard(self):
    """Drop what was written: `close()` commits nothing, whatever is
    written after this. Calling it twice, or after `close()`, does
    nothing more."""
    self._discarded = True
    if not self.closed:
        super().truncate(0)
        super().seek(0)

truncate(size=None)

Source code in src/pathapi/_commit.py
91
92
93
def truncate(self, size=None):
    self._dirty = True
    return super().truncate(size)

write(data)

Source code in src/pathapi/_commit.py
83
84
85
def write(self, data):
    self._dirty = True
    return super().write(data)

writelines(lines)

Source code in src/pathapi/_commit.py
87
88
89
def writelines(self, lines):
    self._dirty = True
    return super().writelines(lines)

DiscardOnError

Mixin for a write stream that commits when it is closed. Leaving a with block through an exception (any BaseException) calls discard() and then closes; leaving it normally just closes. A stream that is garbage collected unclosed while _holds_uncommitted() is true is discarded too, with a ResourceWarning: what was never finished is not committed. Put it first in the bases, so that its __exit__ and __del__ win over io.IOBase's.

__slots__ = () class-attribute instance-attribute

__del__()

Source code in src/pathapi/_commit.py
32
33
34
35
36
37
38
39
40
41
42
43
def __del__(self):
    try:
        if not self.closed and self._holds_uncommitted():
            self.discard()
            _warnings.warn(
                f"unclosed {type(self).__name__} discarded without "
                "committing; close the file",
                ResourceWarning,
                source=self,
            )
    finally:
        super().__del__()

__exit__(exc_type, exc_value, traceback)

Source code in src/pathapi/_commit.py
45
46
47
48
49
50
51
52
def __exit__(self, exc_type, exc_value, traceback):
    try:
        if exc_type is not None:
            self.discard()
    except BaseException:
        super().__exit__(exc_type, exc_value, traceback)
        raise
    return super().__exit__(exc_type, exc_value, traceback)

discard()

Drop what was written so that close() commits nothing. It does nothing more when called twice or after close().

Source code in src/pathapi/_commit.py
21
22
23
24
def discard(self) -> None:
    """Drop what was written so that `close()` commits nothing. It does
    nothing more when called twice or after `close()`."""
    raise NotImplementedError

LRU(func, maxsize=128, on_evict=None)

Bases: Generic[K, V]

Thread-safe memoizing LRU cache over a function, callable like the function itself; invalidate(*args) evicts and recomputes an entry, discard(*args) evicts without recomputing.

on_evict(key, value), when given, is called for every value the cache drops -- LRU overflow, a maxsize shrink, invalidate()/discard(), and the losing result of two concurrent misses on one key -- so a cache of live resources (connections) can close them instead of leaking them. It runs outside the lock, and an exception it raises is suppressed: eviction is a side effect of an unrelated lookup, which must not fail because a stale resource would not close cleanly.

The computation itself runs outside the lock, so concurrent misses on one key may each call func; the first stored result wins and is returned to every caller, and the others are passed to on_evict.

maxsize follows functools.lru_cache: None is unbounded, and 0 (or less) stores nothing, so func runs on every call and its result is returned without reaching on_evict.

Source code in src/pathapi/_lru.py
64
65
66
67
68
69
70
71
72
73
74
def __init__(
    self,
    func: _ty.Callable[K, V],
    maxsize: int | None = 128,
    on_evict: "_ty.Callable[[tuple, V], object] | None" = None,
):
    self.cache = collections.OrderedDict()
    self.func = func
    self._maxsize = maxsize
    self.lock = RLock()
    self.on_evict = on_evict

cache = collections.OrderedDict() instance-attribute

func = func instance-attribute

lock = RLock() instance-attribute

maxsize property writable

on_evict = on_evict instance-attribute

__call__(*args)

Source code in src/pathapi/_lru.py
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
def __call__(self, *args: K.args) -> V:
    cache = self.cache
    with self.lock:
        if args in cache:
            cache.move_to_end(args)
            return cache[args]
    result = self.func(*args)
    if self._maxsize is not None and self._maxsize <= 0:
        return result
    evicted = []
    with self.lock:
        if args in cache:
            # A concurrent miss stored its result first: keep that one,
            # so every caller shares a single value.
            existing = cache[args]
            cache.move_to_end(args)
            if existing is not result:
                evicted.append((args, result))
            result = existing
        else:
            cache[args] = result
            if self._maxsize is not None and len(cache) > self._maxsize:
                evicted.append(cache.popitem(last=False))
    self._dispose(evicted)
    return result

discard(*args)

Drop the entry for args (passing it to on_evict) without recomputing it. Returns whether an entry was present.

Source code in src/pathapi/_lru.py
126
127
128
129
130
131
132
133
134
def discard(self, *args: K.args) -> bool:
    """Drop the entry for `args` (passing it to `on_evict`) without
    recomputing it. Returns whether an entry was present."""
    with self.lock:
        if args not in self.cache:
            return False
        value = self.cache.pop(args)
    self._dispose([(args, value)])
    return True

invalidate(*args)

Source code in src/pathapi/_lru.py
136
137
138
def invalidate(self, *args: K.args) -> V:
    self.discard(*args)
    return self(*args)

OfferedOnce(predicate)

An (error, path) -> bool error predicate that sees each error once.

True swallows the error. An error it answered False for, and an exception it raised itself, are not offered again when they travel up through the enclosing directories of a recursive operation: they are raised as they are. Build one with as_error_predicate().

Source code in src/pathapi/_errorpolicy.py
60
61
62
def __init__(self, predicate: _ty.Callable[[Exception, _ty.Any], bool]) -> None:
    self._predicate = predicate
    self._seen: "list[BaseException]" = []

__slots__ = ('_predicate', '_seen') class-attribute instance-attribute

__call__(error, path)

Source code in src/pathapi/_errorpolicy.py
64
65
66
67
68
69
70
71
72
73
74
75
def __call__(self, error: Exception, path: _ty.Any) -> bool:
    if any(error is seen for seen in self._seen):
        return False
    try:
        swallow = self._predicate(error, path)
    except BaseException as raised:
        self._seen.append(raised)
        raise
    if swallow:
        return True
    self._seen.append(error)
    return False

TreePath

Bases: Path

A Path whose names are POSIX segments below the root of a tree that a backend serves (a remote directory, the members of an archive).

Joined and normalized like PurePosixPath: empty and "." segments collapse and an absolute argument restarts the join. Relative and rooted spellings name the same node, because the tree has no working directory. A subclass supplies the constructor, _rebuild() and the backend it carries; this class holds the names only.

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

parent property

parts property

segments property

__copy__()

Source code in src/pathapi/_tree.py
76
77
def __copy__(self):
    return self._rebuild(self.as_posix())

__deepcopy__(memo)

Source code in src/pathapi/_tree.py
79
80
81
82
def __deepcopy__(self, memo):
    # A backend is a connection or a handle, not part of the value.
    inst = memo[id(self)] = self.__copy__()
    return inst

__str__()

Source code in src/pathapi/_tree.py
73
74
def __str__(self) -> str:
    return self.as_posix()

relative_to(other, *, walk_up=False)

self below other, as a relative path on the same backend.

other is a path of this class on the same tree or a str. Names are compared as written, with no .. applied. ValueError when self is not below other, when the two differ in being rooted, or when the trees differ. With walk_up, a path beside other gets a .. per level climbed; a .. in other cannot be walked.

Source code in src/pathapi/_tree.py
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
def relative_to(self, other, *, walk_up: bool = False):
    """`self` below `other`, as a relative path on the same backend.

    `other` is a path of this class on the same tree or a `str`. Names are
    compared as written, with no `..` applied. `ValueError` when `self`
    is not below `other`, when the two differ in being rooted, or when
    the trees differ. With `walk_up`, a path beside `other` gets a `..`
    per level climbed; a `..` in `other` cannot be walked."""
    if isinstance(other, self._family()):
        if not self._same_filesystem(other):
            raise ValueError(
                f"{str(self)!r} and {str(other)!r} are not in the same tree"
            )
    else:
        other = self.with_segments(other)
    rooted = self.segments[:1] == ("",)
    other_rooted = other.segments[:1] == ("",)
    names = [name for name in self.segments if name]
    other_names = [name for name in other.segments if name]
    for step in range(len(other_names) + 1):
        keep = len(other_names) - step
        if rooted == other_rooted and names[:keep] == other_names[:keep]:
            return self.with_segments(*[".."] * step, *names[keep:])
        if not walk_up:
            raise ValueError(
                f"{str(self)!r} is not in the subpath of {str(other)!r}"
            )
        if keep and other_names[keep - 1] == "..":
            raise ValueError(f"'..' segment in {str(other)!r} cannot be walked")
    raise ValueError(f"{str(self)!r} and {str(other)!r} have different anchors")

with_segments(*segments)

Source code in src/pathapi/_tree.py
104
105
def with_segments(self, *segments: str):
    return self._rebuild("/".join(segments))

as_error_handler(ignore_error, *, default=False)

Normalize an ignore_error argument into a callable error policy.

Every ignore_error (and copy()'s on_error) parameter in this library accepts either a bool or a callable, but the callables have different arities per call site (Path.rm() and Path.copy(on_error=) -> (error, path), Path.copy(ignore_error=) -> (error), PathSyncer.sync() -> (error, source, target, event)). Changing the arity of an existing keyword would break its callers, so this helper only normalizes the bool case and passes a supplied callable through untouched -- it is invoked with whatever arguments its own call site already uses.

None means "no policy supplied": it resolves to default (False for every current caller, i.e. raise on the first error), which preserves Path.copy(ignore_error=None)'s documented meaning.

Centralizing this keeps a fourth call site from drifting back into calling a bool (see PathSyncer.sync()'s symlink branch, which did exactly that and raised TypeError: 'bool' object is not callable).

Source code in src/pathapi/_errorpolicy.py
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def as_error_handler(
    ignore_error: _ty.Union[bool, _ty.Callable[..., bool], None],
    *,
    default: bool = False,
) -> _ty.Callable[..., bool]:
    """Normalize an `ignore_error` argument into a *callable* error policy.

    Every `ignore_error` (and `copy()`'s `on_error`) parameter in this library
    accepts either a bool or a callable, but the callables have **different
    arities** per call site (`Path.rm()` and `Path.copy(on_error=)` ->
    `(error, path)`, `Path.copy(ignore_error=)` -> `(error)`,
    `PathSyncer.sync()` -> `(error, source, target, event)`). Changing the
    arity of an existing keyword would break its callers, so this helper only
    normalizes the *bool* case and passes a supplied callable through
    untouched -- it is invoked with whatever arguments its own call site
    already uses.

    `None` means "no policy supplied": it resolves to `default` (False for
    every current caller, i.e. raise on the first error), which preserves
    `Path.copy(ignore_error=None)`'s documented meaning.

    Centralizing this keeps a fourth call site from drifting back into
    calling a bool (see `PathSyncer.sync()`'s symlink branch, which did
    exactly that and raised `TypeError: 'bool' object is not callable`).
    """
    if callable(ignore_error):
        return ignore_error
    if ignore_error is None:
        ignore_error = default
    result = bool(ignore_error)
    return lambda *args, **kwargs: result

as_error_predicate(on_error)

on_error as an OfferedOnce predicate: a bool becomes a constant answer (None is False), a callable is called as (error, path), and an OfferedOnce is returned as it is, so a recursive call that is handed the predicate of its caller shares what that predicate has seen.

Source code in src/pathapi/_errorpolicy.py
78
79
80
81
82
83
84
85
86
87
def as_error_predicate(
    on_error: _ty.Union[bool, _ty.Callable[[Exception, _ty.Any], bool], None],
) -> OfferedOnce:
    """`on_error` as an `OfferedOnce` predicate: a bool becomes a constant
    answer (`None` is `False`), a callable is called as `(error, path)`, and
    an `OfferedOnce` is returned as it is, so a recursive call that is handed
    the predicate of its caller shares what that predicate has seen."""
    if isinstance(on_error, OfferedOnce):
        return on_error
    return OfferedOnce(as_error_handler(on_error))

check_follow(name, policy)

Raise ValueError naming name unless policy is True, False, None or a callable: what rm() accepts for follow_symlinks= and follow_binds=. An rm() override calls it before removing anything.

Source code in src/pathapi/_errorpolicy.py
90
91
92
93
94
95
def check_follow(name: str, policy) -> None:
    """Raise `ValueError` naming `name` unless `policy` is `True`, `False`,
    `None` or a callable: what `rm()` accepts for `follow_symlinks=` and
    `follow_binds=`. An `rm()` override calls it before removing anything."""
    if not (policy is None or callable(policy) or isinstance(policy, bool)):
        raise ValueError(f"{name} must be True, False, None or a callable")

contains(src, target)

Whether target is src or lies inside it (pathlib 3.14's copy() refuses both), looking through symlinks. Paths _same_filesystem() reports in different namespaces are never nested.

Source code in src/pathapi/protocols/_relation.py
152
153
154
155
156
157
158
159
160
def contains(src: Path, target: Path) -> bool:
    """Whether `target` is `src` or lies inside it (pathlib 3.14's copy()
    refuses both), looking through symlinks. Paths `_same_filesystem()`
    reports in different namespaces are never nested."""
    return _relation(src, target, follow=True, identity=False) in (
        _SAME,
        _OTHER,
        _INNER,
    )

copy_error_predicate(ignore_error, on_error)

The (error, path) -> bool predicate for copy()'s on_error=, or None when it was not given. Raises TypeError when ignore_error= was given too: a copy() override calls this before it copies anything.

Source code in src/pathapi/_errorpolicy.py
 98
 99
100
101
102
103
104
105
106
def copy_error_predicate(ignore_error, on_error) -> "OfferedOnce | None":
    """The `(error, path) -> bool` predicate for `copy()`'s `on_error=`, or
    None when it was not given. Raises `TypeError` when `ignore_error=` was
    given too: a `copy()` override calls this before it copies anything."""
    if on_error is None:
        return None
    if ignore_error is not None:
        raise TypeError("copy() takes ignore_error or on_error, not both")
    return as_error_predicate(on_error)

derived_from(*primitives)

Mark a method as built on other primitives of the class, such as an iterdir() that lists through _listdir(): protocols.supports() counts it as implemented only when one of those primitives is.

Source code in src/pathapi/_stubs.py
49
50
51
52
53
54
55
56
57
58
def derived_from(*primitives: str):
    """Mark a method as built on other primitives of the class, such as an
    `iterdir()` that lists through `_listdir()`: `protocols.supports()` counts
    it as implemented only when one of those primitives is."""

    def mark(function):
        setattr(function, DERIVED_ATTR, primitives)
        return function

    return mark

is_safe_child_name(name, *, windows=False)

Whether name is a single path component that stays inside its parent.

Use this before joining a name that came from somewhere untrusted (a remote listing, an archive member, a sync source) onto a destination. Always rejected: a non-str, "", ".", "..", and any name containing / or NUL. With windows=True (see is_windows_flavoured) also rejected: \, : (a drive prefix such as D:x, or an NTFS alternate data stream), a trailing dot or space (Windows drops it, so "report." and "report" name one file), and the reserved device names CON, PRN, AUX, NUL, COM1-COM9 and LPT1-LPT9, in any case and with or without an extension ("nul.txt").

Source code in src/pathapi/_names.py
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def is_safe_child_name(name: object, *, windows: bool = False) -> bool:
    """Whether `name` is a single path component that stays inside its parent.

    Use this before joining a name that came from somewhere untrusted (a
    remote listing, an archive member, a sync source) onto a destination.
    Always rejected: a non-`str`, `""`, `"."`, `".."`, and any name
    containing `/` or NUL. With `windows=True` (see `is_windows_flavoured`)
    also rejected: `\\`, `:` (a drive prefix such as `D:x`, or an NTFS
    alternate data stream), a trailing dot or space (Windows drops it, so
    `"report."` and `"report"` name one file), and the reserved device names
    `CON`, `PRN`, `AUX`, `NUL`, `COM1`-`COM9` and `LPT1`-`LPT9`, in any case
    and with or without an extension (`"nul.txt"`).
    """
    return _is_safe_name(name, windows, True)

is_windows_flavoured(path)

Whether names joined onto path are interpreted with Windows rules.

True for a pathlib.PureWindowsPath (so LocalPath on Windows) and for a path whose filepath is one (a file: URI path on Windows). Everything else -- MemPath, remote schemes, POSIX local paths -- treats \ and : as ordinary filename characters.

Source code in src/pathapi/_names.py
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
def is_windows_flavoured(path: object) -> bool:
    """Whether names joined onto `path` are interpreted with Windows rules.

    True for a `pathlib.PureWindowsPath` (so `LocalPath` on Windows) and for
    a path whose `filepath` is one (a `file:` URI path on Windows). Everything else --
    `MemPath`, remote schemes, POSIX local paths -- treats `\\` and `:` as
    ordinary filename characters.
    """
    if isinstance(path, _PureWindowsPath):
        return True
    try:
        filepath = getattr(path, "filepath", None)
    except Exception:
        return False
    return isinstance(filepath, _PureWindowsPath)

name_stem(name)

Source code in src/pathapi/path/_pure.py
67
68
69
def _name_stem(name: str) -> str:
    i = name.rfind(".")
    return name[:i] if 0 < i < len(name) - 1 else name

name_suffix(name)

Source code in src/pathapi/path/_pure.py
63
64
65
def _name_suffix(name: str) -> str:
    i = name.rfind(".")
    return name[i:] if 0 < i < len(name) - 1 else ""

name_suffixes(name)

Source code in src/pathapi/path/_pure.py
71
72
73
74
def _name_suffixes(name: str) -> "list[str]":
    if name.endswith("."):
        return []
    return ["." + suffix for suffix in name.lstrip(".").split(".")[1:]]

notimplemented(method)

Mark a method as the stub of a primitive. Used bare (@notimplemented) or with the reason a backend gives (@notimplemented("why")); calling the stub raises NotImplementedError with the reason, or with the method's name.

Source code in src/pathapi/_stubs.py
23
24
25
26
27
28
29
30
31
def notimplemented(method):
    """Mark a method as the stub of a primitive. Used bare (`@notimplemented`)
    or with the reason a backend gives (`@notimplemented("why")`); calling the
    stub raises `NotImplementedError` with the reason, or with the method's
    name."""
    if isinstance(method, str):
        reason = method
        return lambda function: _stub(function, reason)
    return _stub(method, None)

parse_mode(mode)

Normalize a permission mode to an int, parsing str as octal.

chmod("0755") is the spelling everyone actually writes a mode in -- chmod(1), Ansible, Dockerfiles, every shell script -- and stdlib refuses it (TypeError: 'str' object cannot be interpreted as an integer). This library accepts it, which makes the base explicit and non-negotiable rather than leaving it to each call site.

Why base 8 is mandatory here, and never a plain int(): "0755" parsed as decimal is 755, which is 0o1363 -- a different and valid mode. Nothing would raise; the file would just end up with permissions nobody intended. That is exactly why stdlib declines strings, so the only safe way to accept them is to parse them one way, in one place.

Accepts an optional 0o/0O prefix. Anything outside [0-7] raises ValueError rather than being coerced -- a mode is not a number that happens to be written in octal, it is octal.

An int passes through (including 0o755, which is an int by the time it gets here -- the literal is resolved by the parser, so chmod(0o755) and chmod("0755") agree). A bool is TypeError, and a number that cannot be a mode (negative, or wider than the 16 bits of a st_mode) is ValueError, for an int and a str alike. The file-type bits of a st_mode are in range, as os.chmod accepts them; stat.S_IMODE() drops them.

Source code in src/pathapi/_permissions.py
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def parse_mode(mode: _ty.Union[int, str]) -> int:
    """Normalize a permission `mode` to an int, parsing `str` as **octal**.

    `chmod("0755")` is the spelling everyone actually writes a mode in --
    `chmod(1)`, Ansible, Dockerfiles, every shell script -- and stdlib
    refuses it (`TypeError: 'str' object cannot be interpreted as an
    integer`). This library accepts it, which makes the base explicit and
    non-negotiable rather than leaving it to each call site.

    **Why base 8 is mandatory here, and never a plain `int()`:** `"0755"`
    parsed as decimal is 755, which is `0o1363` -- a different *and valid*
    mode. Nothing would raise; the file would just end up with permissions
    nobody intended. That is exactly why stdlib declines strings, so the
    only safe way to accept them is to parse them one way, in one place.

    Accepts an optional `0o`/`0O` prefix. Anything outside `[0-7]` raises
    `ValueError` rather than being coerced -- a mode is not a number that
    happens to be written in octal, it is octal.

    An `int` passes through (including `0o755`, which *is* an int by the time
    it gets here -- the literal is resolved by the parser, so `chmod(0o755)`
    and `chmod("0755")` agree). A `bool` is `TypeError`, and a number that
    cannot be a mode (negative, or wider than the 16 bits of a `st_mode`) is
    `ValueError`, for an `int` and a `str` alike. The file-type bits of a
    `st_mode` are in range, as `os.chmod` accepts them; `stat.S_IMODE()` drops
    them.
    """
    if isinstance(mode, bool):
        raise TypeError("a mode is an int or an octal str, not bool")
    if isinstance(mode, str):
        text = mode.strip()
        if text[:2].lower() == "0o":
            text = text[2:]
        if not text or any(character not in "01234567" for character in text):
            raise ValueError(f"invalid octal mode: {mode!r}")
        value = int(text, 8)
    else:
        value = _operator.index(mode)
    if not 0 <= value <= _MODE_MASK:
        raise ValueError(f"mode out of range (0 to {_MODE_MASK:#o}): {mode!r}")
    return value

parse_owner(uid, gid)

Normalize a chown() uid/gid pair to canonical int | None.

None means "leave unchanged". -1 is accepted as an alias for it, since that is how os.chown spells the same thing and callers arriving from the stdlib reach for it out of habit.

The point of centralizing this is that every backend spells "unchanged" differently -- os.chown wants -1, SFTP setstat wants the field omitted from the attrs entirely, and other middlewares want None. If each scheme translated the caller's input itself, that is three chances for the semantics to disagree. Normalizing once on Path means a backend's _chown() receives an already-canonical pair and only has to convert to its own wire spelling.

A str is passed through as a name (shutil.chown accepts user and group names, and it is useful not to force a caller to resolve them) -- backends that cannot resolve names should say so rather than guess.

Source code in src/pathapi/_permissions.py
 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
def parse_owner(
    uid: _ty.Union[int, str, None], gid: _ty.Union[int, str, None]
) -> "_ty.Tuple[_ty.Optional[int], _ty.Optional[int]]":
    """Normalize a `chown()` uid/gid pair to canonical `int | None`.

    `None` means "leave unchanged". `-1` is accepted as an alias for it,
    since that is how `os.chown` spells the same thing and callers arriving
    from the stdlib reach for it out of habit.

    The point of centralizing this is that **every backend spells
    "unchanged" differently** -- `os.chown` wants `-1`, SFTP `setstat` wants
    the field omitted from the attrs entirely, and other middlewares want
    `None`. If each scheme translated the caller's input itself, that is
    three chances for the semantics to disagree. Normalizing once on `Path`
    means a backend's `_chown()` receives an already-canonical pair and only
    has to convert to its own wire spelling.

    A `str` is passed through as a *name* (`shutil.chown` accepts user and
    group names, and it is useful not to force a caller to resolve them) --
    backends that cannot resolve names should say so rather than guess.
    """

    def _one(value):
        if value is None or isinstance(value, str):
            return value
        if isinstance(value, bool):
            raise TypeError("an owner is an id, a name or None, not bool")
        value = _operator.index(value)
        if value == -1:
            return None
        if not 0 <= value <= _ID_MAX:
            raise ValueError(f"id out of range (0 to {_ID_MAX}): {value!r}")
        return value

    return _one(uid), _one(gid)

primitive(method)

Mark method as the default of a primitive: it raises NotImplementedError, and a class that does not replace it does not implement the primitive.

Source code in src/pathapi/_stubs.py
15
16
17
18
19
20
def primitive(method):
    """Mark `method` as the default of a primitive: it raises
    `NotImplementedError`, and a class that does not replace it does not
    implement the primitive."""
    setattr(method, STUB_ATTR, True)
    return method