Skip to content

Utilities API

pathlib_next.utils.glob

full_match(segments, pattern, case_sensitive)

Match segments against a glob pattern that may contain "**" components matching zero or more segments (pathlib 3.13's PurePath.full_match semantics).

Source code in src/pathlib_next/utils/glob.py
30
31
32
33
34
def full_match(segments: _ty.Sequence[str], pattern: str, case_sensitive: bool) -> bool:
    """Match `segments` against a glob pattern that may contain "**"
    components matching zero or more segments (pathlib 3.13's
    PurePath.full_match semantics)."""
    return _full_match(tuple(segments), tuple(pattern.split("/")), case_sensitive)

glob(path, *, dironly=False, root_dir=None, recursive=False, include_hidden=False, case_sensitive=None)

Return an iterator which yields the paths matching a pathname pattern.

The pattern may contain simple shell-style wildcards a la fnmatch. However, unlike fnmatch, filenames starting with a dot are special cases that are not matched by '*' and '?' patterns.

If recursive is true, the pattern '**' will match any files and zero or more directories and subdirectories.

Source code in src/pathlib_next/utils/glob.py
 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
 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
def glob(
    path: _Globable,
    *,
    dironly: bool = False,
    root_dir: _Globable | None = None,
    recursive: bool = False,
    include_hidden: bool = False,
    case_sensitive: bool | None = None,
) -> _ty.Iterable[_Globable]:
    """Return an iterator which yields the paths matching a pathname pattern.

    The pattern may contain simple shell-style wildcards a la
    fnmatch. However, unlike fnmatch, filenames starting with a
    dot are special cases that are not matched by '*' and '?'
    patterns.

    If recursive is true, the pattern '**' will match any files and
    zero or more directories and subdirectories.
    """
    if case_sensitive is None:
        case_sensitive = path._is_case_sensitive

    include_hidden = include_hidden or path.is_hidden()
    pattern = compile_pattern(path.name, case_sensitive) if path.name else ANY_PATTERN

    name_is_pattern = WILDCARD_PATTERN.search(path.name) != None
    wildcard_in_path = name_is_pattern or path.has_glob_pattern()
    parent = next(iter(path.parents), None)

    root: _Globable = (
        (root_dir or parent) if not root_dir or not parent else (root_dir / parent)
    )

    if recursive and path.name == RECURSIVE:
        globber = _glob_recursive
    else:
        globber = _glob_with_pattern

    if not parent or not wildcard_in_path:
        yield from globber(
            root or path,
            pattern,
            dironly,
            include_hidden=include_hidden,
        )
        return

    # Recurse to enumerate matching directories only when the *parent*
    # portion itself contains a wildcard (e.g. "sub*/*.py" needs every
    # "sub*"-matching dir found first). Using `name_is_pattern` (whether the
    # *leaf* is a pattern) here instead -- which is almost always true, since
    # that's the common case of a literal directory + wildcarded filename --
    # was wrong: it re-globbed `parent`'s own parent to "rediscover" parent
    # by name, which only degenerates back to plain `[parent]` when `parent`
    # has a non-empty literal name to match against. It silently returned
    # the wrong directory set when `parent.name` is "" (MemPath's virtual
    # root, and -- untested here, LocalPath at an OS filesystem root).
    if parent and parent.has_glob_pattern():
        dirs = glob(
            parent,
            root_dir=root_dir,
            recursive=recursive,
            dironly=True,
            include_hidden=include_hidden,
            case_sensitive=case_sensitive,
        )
    else:
        dirs = [parent]

    for parent in dirs:
        yield from globber(parent, pattern, dironly, include_hidden)

pathlib_next.utils.stat

FileStat(st_mode=None, st_size=0, st_mtime=0, is_dir=False)

Bases: FileStatLike

Concrete, slotted FileStatLike for backends without a real os.stat_result (e.g. MemPath, HttpPath). from_path() builds one from any object with a stat() method, or passes a FileStat through unchanged.

Source code in src/pathlib_next/utils/stat.py
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def __init__(
    self,
    st_mode: int = None,
    st_size: int = 0,
    st_mtime: int = 0,
    is_dir: bool = False,
):
    self.st_mode = st_mode or (
        _stat.S_IFDIR | 0o555 if is_dir else _stat.S_IFREG | 0o444
    )
    self.st_nlink = 1
    self.st_uid = 0
    self.st_gid = 0
    self.st_size = st_size
    self.st_atime = 0
    self.st_mtime = st_mtime
    self.st_ctime = 0

from_stat(stat) classmethod

Copy any stat-like object's (os.stat_result, paramiko's SFTPAttributes, ...) recognized fields into a fresh FileStat, so downstream code (e.g. .is_dir()) can rely on a uniform type. Passes an already-FileStat through unchanged.

Source code in src/pathlib_next/utils/stat.py
80
81
82
83
84
85
86
87
88
89
90
91
@classmethod
def from_stat(cls, stat: _ty.Any) -> "FileStat":
    """Copy any stat-like object's (`os.stat_result`, paramiko's
    `SFTPAttributes`, ...) recognized fields into a fresh `FileStat`,
    so downstream code (e.g. `.is_dir()`) can rely on a uniform type.
    Passes an already-`FileStat` through unchanged."""
    if isinstance(stat, FileStat):
        return stat
    result = FileStat.__new__(FileStat)
    for prop in FileStat.__slots__:
        setattr(result, prop, getattr(stat, prop, 0))
    return result

is_block_device()

Whether this path is a block device.

Source code in src/pathlib_next/utils/stat.py
120
121
122
123
124
def is_block_device(self):
    """
    Whether this path is a block device.
    """
    return _stat.S_ISBLK(self.st_mode)

is_char_device()

Whether this path is a character device.

Source code in src/pathlib_next/utils/stat.py
126
127
128
129
130
def is_char_device(self):
    """
    Whether this path is a character device.
    """
    return _stat.S_ISCHR(self.st_mode)

is_dir()

Whether this path is a directory.

Source code in src/pathlib_next/utils/stat.py
101
102
103
104
105
def is_dir(self):
    """
    Whether this path is a directory.
    """
    return _stat.S_ISDIR(self.st_mode)

is_fifo()

Whether this path is a FIFO.

Source code in src/pathlib_next/utils/stat.py
132
133
134
135
136
def is_fifo(self):
    """
    Whether this path is a FIFO.
    """
    return _stat.S_ISFIFO(self.st_mode)

is_file()

Whether this path is a regular file (also True for symlinks pointing to regular files).

Source code in src/pathlib_next/utils/stat.py
107
108
109
110
111
112
def is_file(self):
    """
    Whether this path is a regular file (also True for symlinks pointing
    to regular files).
    """
    return _stat.S_ISREG(self.st_mode)

is_socket()

Whether this path is a socket.

Source code in src/pathlib_next/utils/stat.py
138
139
140
141
142
def is_socket(self):
    """
    Whether this path is a socket.
    """
    return _stat.S_ISSOCK(self.st_mode)

Whether this path is a symbolic link.

Source code in src/pathlib_next/utils/stat.py
114
115
116
117
118
def is_symlink(self):
    """
    Whether this path is a symbolic link.
    """
    return _stat.S_ISLNK(self.st_mode)

pathlib_next.utils.sync

PathSyncer(checksum=None, /, remove_missing=False, follow_symlinks=True, symlink_mode='preserve', hook=None, ignore_error=False, quick_check=True)

Bases: object

One-way checksum-driven tree sync: copies/creates in target whatever differs from source (by checksum), optionally removing files in target that are missing from source. Works across any two Path implementations (e.g. MemPath -> LocalPath, or between two UriPath schemes) -- see sync().

The default checksum policy prefers each side's backend-native digest (protocols.checksum.NativeChecksum.checksum(), e.g. SftpPath's check-file@openssh.com support) over streaming the file through open("rb"), but only when BOTH sides can produce a digest under the same algorithm -- native or streamed. If either side can't (missing the protocol, or it raises NotImplementedError for the requested algorithm), both sides fall back to streaming rather than comparing a native digest to a streamed one. A custom checksum callable disables this native-preferring behavior entirely (it is called exactly as before, once per side, compared with ==).

quick_check=True (the default) adds a cheap metadata-only pre-check -- the classic rsync "quick check" heuristic -- for any pair where at least one side is non-local (Uri.is_local(); a side without an is_local() method at all, e.g. plain LocalPath/MemPath, is treated as local): if st_size AND st_mtime already match (from the listing/stat metadata PathAndStat already carries -- no extra round trip), the pair is treated as in sync WITHOUT calling checksum at all, native or streamed. A mismatch on either falls through to a real checksum comparison rather than being treated as "changed" -- mtime can be unreliable across backends/clock skew, so a false "needs copy" from a mismatch is merely wasteful, while a false "in sync" would be a correctness regression. Local-to-local pairs always skip this pre-check (unchanged pre-existing behavior -- local reads are already cheap, and this project's copy(preserve_metadata=True) doesn't guarantee mtime propagation on every path, see docs/divergences.md). Set quick_check=False to disable the pre-check entirely and always checksum, matching pre-quick_check behavior for non-local pairs too.

follow_symlinks (default True) controls whether a symlink source is resolved during traversal (content synced as if it weren't a link) or reported as a symlink (is_symlink() true). When it's False and a symlink source is reached, symlink_mode decides what happens: "preserve" (default) creates a matching symlink on target with the same raw, unresolved target string readlink() returned (dangling links and relative targets included -- never resolved/validated); "reject" raises NotImplementedError instead (the only behavior before this kwarg existed). If target can't create symlinks at all (most backends -- only LocalPath and SftpPath currently implement symlink_to()), "preserve" mode raises NotImplementedError too, through the same ignore_error/hook() machinery as every other branch.

Source code in src/pathlib_next/utils/sync.py
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
def __init__(
    self,
    checksum: _ty.Callable[[PathAndStat], _ty.Any] | None = None,
    /,
    remove_missing: bool = False,
    follow_symlinks: bool = True,
    symlink_mode: '_ty.Literal["preserve", "reject"]' = "preserve",
    hook: _ty.Callable[[PathAndStat, PathAndStat, SyncEvent, bool], None] = None,
    ignore_error: _OnPathSyncerError | bool = False,
    quick_check: bool = True,
) -> None:
    # `None` (the default) resolves to `_default_checksum` -- a sentinel
    # `sync()` recognizes (via `is`) to route through
    # `_default_checksums_match()` instead of two independent calls, so
    # the native-vs-streaming decision can be coordinated across BOTH
    # sides at once. A caller-supplied callable is stored and used
    # as-is (`checksum(target) == checksum(source)`, unchanged from
    # before this feature).
    if checksum is None:
        checksum = _default_checksum
    self.checksum = checksum
    self.remove_missing = remove_missing
    self._hook = hook
    self.follow_symlinks = follow_symlinks
    if symlink_mode not in ("preserve", "reject"):
        raise ValueError(
            f"symlink_mode must be 'preserve' or 'reject', got {symlink_mode!r}"
        )
    self.symlink_mode = symlink_mode
    self.ignore_error = _ty.cast(
        _OnPathSyncerError, _utils.as_error_handler(ignore_error)
    )
    self.quick_check = quick_check

sync(source, target, /, dry_run=False, ignore_error=None)

Sync source onto target.

ignore_error overrides the instance-level policy for this call only. It accepts a bool or a callable with the same (error, source, target, event) arity as the constructor's; None (the default) means "use the policy given to __init__".

The default used to be the bool False, which both shadowed a constructor-supplied policy and was called directly by the symlink branch (TypeError: 'bool' object is not callable). Passing a callable explicitly behaves exactly as before.

Source code in src/pathlib_next/utils/sync.py
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
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
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
def sync(
    self,
    source: Path | PathAndStat,
    target: Path | PathAndStat,
    /,
    dry_run: bool = False,
    ignore_error: _OnPathSyncerError | bool | None = None,
):
    """Sync `source` onto `target`.

    `ignore_error` overrides the instance-level policy for this call
    only. It accepts a bool or a callable with the same
    `(error, source, target, event)` arity as the constructor's; `None`
    (the default) means "use the policy given to `__init__`".

    The default used to be the bool `False`, which both shadowed a
    constructor-supplied policy and was *called* directly by the symlink
    branch (`TypeError: 'bool' object is not callable`). Passing a
    callable explicitly behaves exactly as before.
    """
    checksum = self.checksum
    _ignore_error = (
        self.ignore_error
        if ignore_error is None
        else _utils.as_error_handler(ignore_error)
    )

    def start():
        nonlocal source, target
        source = (
            PathAndStat(source, follow_symlink=self.follow_symlinks)
            if not isinstance(source, PathAndStat)
            else source
        )
        target = (
            PathAndStat(target, follow_symlink=self.follow_symlinks)
            if not isinstance(target, PathAndStat)
            else target
        )

    if self.hook(source, target, SyncEvent.SyncStart, False, start, _ignore_error):
        return

    if not source.exists():
        if self.remove_missing:
            if self.hook(
                source,
                target,
                SyncEvent.RemovedMissing,
                dry_run,
                lambda: target.path.rm(recursive=True, missing_ok=True),
                _ignore_error,
            ):
                return
    elif source.is_symlink():
        if self.symlink_mode == "reject":
            error = NotImplementedError("symlink sync not implemented yet")
            if not _ignore_error(error, source, target, SyncEvent.Symlink):
                raise error
            return

        def create_symlink():
            # Raw, unresolved target string -- readlink() returns a
            # Path-like object on every implementation that has it
            # (stdlib Path, or SftpPath's `with_segments(target)`, a
            # Uri carrying the *source's* host/scheme). Uri.as_posix()
            # prepends "host:" (or "user@host:") when a source/host is
            # present, which corrupts a bare relative/absolute symlink
            # target (e.g. "real.txt" -> "host:real.txt") -- so Uri's
            # own `.path` (the raw, un-prefixed path string, no host)
            # is used when available; plain stdlib Path has no `.path`
            # attribute at all, so `.as_posix()` is the correct and
            # only accessor there. Never resolved against source's
            # parent -- a relative target stays relative either way.
            link = source.path.readlink()
            raw_target = link.path if hasattr(link, "path") else link.as_posix()

            # Type mismatch: target exists as something other than a
            # symlink (file or dir) -- clear it first, same pattern as
            # the Copy branch above.
            if target.is_file() or target.is_symlink():
                target.path.unlink()
            elif target.exists():
                target.path.rm(recursive=target.is_dir())

            symlink_to = getattr(target.path, "symlink_to", None)
            if symlink_to is None:
                # target backend has no symlink_to() at all (e.g.
                # MemPath, HttpPath) -- normalize to the same
                # NotImplementedError reject mode raises, so
                # ignore_error/hook() callers see one consistent
                # error shape regardless of *why* symlink creation
                # isn't possible. A backend that DOES define
                # symlink_to() but itself raises NotImplementedError
                # (e.g. a future @_utils.notimplemented stub) is left
                # to propagate its own error unchanged -- only the
                # "attribute doesn't exist at all" case is normalized
                # here.
                raise NotImplementedError(
                    "symlink_to() not supported by " f"{type(target.path).__name__}"
                )
            symlink_to(raw_target)

        if self.hook(
            source,
            target,
            SyncEvent.Symlink,
            dry_run,
            create_symlink,
            _ignore_error,
        ):
            return
    elif source.is_file():
        synced = False
        if target.is_file():
            # quick_check: cheap metadata-only pre-check for non-local
            # pairs (see class docstring) -- a match skips checksumming
            # entirely; a mismatch always falls through to a real
            # checksum comparison, never concludes "changed" on its
            # own.
            quick_matched = (
                self.quick_check
                and (not _is_local(source.path) or not _is_local(target.path))
                and _quick_check_in_sync(source, target)
            )
            if quick_matched:
                matches = True
            elif checksum is _default_checksum:
                # Route through the paired native-vs-streaming policy
                # (see class docstring) instead of two independent
                # single-path calls -- only this branch can coordinate
                # "both native or both streamed" across both sides.
                matches = _default_checksums_match(source, target)
            else:
                matches = checksum(target) == checksum(source)
            if matches:
                synced = True
        if not synced:

            def copy():
                if target.is_file() or target.is_symlink():
                    target.path.unlink()
                else:
                    if target.exists():
                        target.path.rm(recursive=target.is_dir())
                source.path.copy(target.path)

            if self.hook(
                source, target, SyncEvent.Copy, dry_run, copy, _ignore_error
            ):
                return
    else:
        if target.is_file():
            if self.hook(
                source,
                target,
                SyncEvent.TypeMismatch,
                dry_run,
                lambda: target.path.unlink(),
                _ignore_error,
            ):
                return

            target._stat = None

        if not target.exists():
            if self.hook(
                source,
                target,
                SyncEvent.CreatedDirectory,
                dry_run,
                lambda: target.path.mkdir(),
                _ignore_error,
            ):
                return

        source_children = None

        def get_source_children():
            nonlocal source_children
            if source_children is None:
                source_children = self._children(source)
            return source_children

        if self.remove_missing:

            def checkchildren():
                source_names = {child.path.name for child in get_source_children()}
                for child in self._children(target):

                    def checkchild():
                        if child.path.name not in source_names:
                            self.hook(
                                source,
                                target,
                                SyncEvent.RemovedMissing,
                                dry_run,
                                lambda child=child: child.path.rm(recursive=True),
                                _ignore_error,
                            )

                    self.hook(
                        source,
                        target,
                        SyncEvent.CheckTargetChild,
                        False,
                        checkchild,
                        _ignore_error,
                    )

            self.hook(
                source,
                target,
                SyncEvent.CheckTargetChildren,
                False,
                checkchildren,
                _ignore_error,
            )

        def sync_children():
            for child in get_source_children():
                self.hook(
                    source,
                    target,
                    SyncEvent.SyncChild,
                    False,
                    # Propagate the resolved policy into the recursive
                    # call so a per-call override applies to the whole
                    # subtree, not just this level.
                    lambda child=child: self.sync(
                        child,
                        target.path / (child.path.name or child.path.parent.name),
                        dry_run,
                        _ignore_error,
                    ),
                    _ignore_error,
                )

        self.hook(
            source,
            target,
            SyncEvent.SyncChildren,
            False,
            sync_children,
            _ignore_error,
        )

    self.hook(source, target, SyncEvent.Synced, dry_run, None, _ignore_error)

SyncEvent

Bases: Enum

Events PathSyncer.hook() fires during a sync, for progress/logging callbacks.

pathlib_next.utils

LRU(func, maxsize=128)

Bases: Generic[K, V]

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

Source code in src/pathlib_next/utils/__init__.py
42
43
44
45
46
def __init__(self, func: _ty.Callable[K, V], maxsize=128):
    self.cache = collections.OrderedDict()
    self.func = func
    self._maxsize = maxsize
    self.lock = RLock()

as_error_handler(ignore_error, *, default=False)

Normalize an ignore_error argument into a callable error policy.

Every ignore_error parameter in this library accepts either a bool or a callable, but the callables have deliberately different arities per call site (Path.rm() -> (error, path), Path.copy() -> (error), PathSyncer.sync() -> (error, source, target, event)). Unifying those arities would break existing 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/pathlib_next/utils/__init__.py
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
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` parameter in this library accepts either a bool or
    a callable, but the callables have **deliberately different arities**
    per call site (`Path.rm()` -> `(error, path)`, `Path.copy()` ->
    `(error)`, `PathSyncer.sync()` -> `(error, source, target, event)`).
    Unifying those arities would break existing 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_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 untouched (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).

Source code in src/pathlib_next/utils/__init__.py
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
def as_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 untouched (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).
    """
    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}")
        return int(text, 8)
    return _operator.index(mode)

as_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/pathlib_next/utils/__init__.py
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
def as_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
        value = _operator.index(value)
        return None if value == -1 else value

    return _one(uid), _one(gid)