Utilities API

`pathlib_next.utils`

`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.

Source code in src/pathlib_next/utils/__init__.py

def __init__(
    self,
    func: _ty.Callable[K, V],
    maxsize=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

`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/pathlib_next/utils/__init__.py

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

`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

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

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

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)

`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), and names that are empty or ./.. once Windows strips trailing dots and spaces (".. ").

Source code in src/pathlib_next/utils/__init__.py

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), and names that are empty or `.`/`..` once
    Windows strips trailing dots and spaces (`".. "`).
    """
    if not isinstance(name, str) or name in ("", ".", "..") or "/" in name:
        return False
    if "\0" in name:
        return False
    if windows:
        if "\\" in name or ":" in name:
            return False
        if name.rstrip(" .") == "":
            return False
    return 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 (FileUri on Windows). Everything else -- MemPath, remote schemes, POSIX local paths -- treats \ and : as ordinary filename characters.

Source code in src/pathlib_next/utils/__init__.py

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 (`FileUri` 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)

`parsedate(date)`

Convert a date to UTC epoch seconds, independent of the host timezone.

str (RFC 1123, RFC 850 or asctime, as in Last-Modified): parsed with email.utils.parsedate_tz, and its zone offset applied. A string with no zone is read as UTC -- HTTP dates are GMT by definition.
int/float: already epoch seconds, returned unchanged.
struct_time/tuple: read as UTC (calendar.timegm), minus its tm_gmtoff (or a parsedate_tz-style 10th element) when one is set.
None or an unparseable string: 0.

Local time (time.mktime) is never involved: it shifted every HTTP/DAV st_mtime by the host's UTC offset and overflowed on Windows for dates near the epoch.

Source code in src/pathlib_next/utils/__init__.py

def parsedate(date: _ty.Union[str, _time.struct_time, tuple, int, float, None]):
    """Convert a date to UTC epoch seconds, independent of the host timezone.

    - `str` (RFC 1123, RFC 850 or asctime, as in `Last-Modified`): parsed
      with `email.utils.parsedate_tz`, and its zone offset applied. A string
      with no zone is read as UTC -- HTTP dates are GMT by definition.
    - `int`/`float`: already epoch seconds, returned unchanged.
    - `struct_time`/tuple: read as UTC (`calendar.timegm`), minus its
      `tm_gmtoff` (or a `parsedate_tz`-style 10th element) when one is set.
    - `None` or an unparseable string: `0`.

    Local time (`time.mktime`) is never involved: it shifted every HTTP/DAV
    `st_mtime` by the host's UTC offset and overflowed on Windows for dates
    near the epoch.
    """
    # Missing/unparseable dates yield epoch 0, not "now" -- a caller with no
    # Last-Modified header shouldn't have that read as "just modified" and
    # poison checksum/sync freshness comparisons.
    if date is None:
        return 0
    if isinstance(date, (int, float)):
        return date
    if isinstance(date, str):
        try:
            parsed = _parsedate_tz(date)
            if parsed is None:
                return 0
            return _calendar.timegm(parsed[:6]) - (parsed[9] or 0)
        except (TypeError, ValueError, IndexError, OverflowError):
            return 0
    offset = getattr(date, "tm_gmtoff", None)
    if offset is None and len(date) > 9:
        offset = date[9]
    return _calendar.timegm(tuple(date)[:6]) - (offset or 0)

`pathlib_next.utils.glob`

Filename globbing over any pathlib_next Path.

Modelled on CPython's pathlib glob selectors, reworked to run over anything that implements _scandir()/iterdir()/is_dir()/name like pathlib.Path (LocalPath, MemPath, every UriPath scheme).

`NonRelativePatternError`

Bases: NotImplementedError, ValueError

An absolute or anchored glob pattern. pathlib raises NotImplementedError("Non-relative patterns are unsupported"); this is also a ValueError, so either except clause catches it.

`full_match(segments, pattern, case_sensitive)`

Match segments against a glob pattern that may contain "" components (pathlib 3.13's PurePath.full_match semantics): a "" matches zero or more segments, except a trailing "" after other components, which needs at least one ("a/" does not match "a").

Runs as a set-of-states automaton, O(len(segments) * len(pattern)), so repeated "**" cannot backtrack exponentially.

Source code in src/pathlib_next/utils/glob.py

def full_match(segments: _ty.Sequence[str], pattern: str, case_sensitive: bool) -> bool:
    """Match `segments` against a glob pattern that may contain "**"
    components (pathlib 3.13's PurePath.full_match semantics): a "**" matches
    zero or more segments, except a trailing "**" after other components,
    which needs at least one ("a/**" does not match "a").

    Runs as a set-of-states automaton, O(len(segments) * len(pattern)), so
    repeated "**" cannot backtrack exponentially.
    """
    pats = pattern.split("/")
    pats = _collapse_recursive(p for i, p in enumerate(pats) if p or i == 0)
    segs = [s for i, s in enumerate(segments) if s or i == 0]
    end = len(pats)

    def closure(states: _ty.Set[int]) -> _ty.Set[int]:
        for i in sorted(states):
            # A non-trailing (or sole) "**" may match zero segments.
            if i < end and pats[i] == RECURSIVE and (i < end - 1 or end == 1):
                states.add(i + 1)
        return states

    states = closure({0})
    for seg in segs:
        advanced: _ty.Set[int] = set()
        for i in states:
            if i == end:
                continue
            pat = pats[i]
            if pat == RECURSIVE:
                advanced.update((i, i + 1))
            elif compile_pattern(pat, case_sensitive).match(seg):
                advanced.add(i + 1)
        if not advanced:
            return False
        states = closure(advanced)
    return end in states

`glob(path, *, dironly=False, root_dir=None, recursive=False, include_hidden=False, case_sensitive=None, native=True, on_error=None, bound_loops=False)`

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

path is the pattern itself, as a path (e.g. UriPath("file:/x/**/*.py")). The pattern may contain simple shell-style wildcards a la fnmatch. Like the stdlib glob module, and unlike Path.glob(), hidden entries (names starting with a dot) are not matched by wildcards and not descended into by "**" unless include_hidden is true.

If recursive is true, the pattern '' will match any files and zero or more directories and subdirectories. recursive=None decides from the pattern itself, as Path.glob() does: a "" component enables it. native= follows the running interpreter's rules, or applies one rule on every version -- see parse_pattern().

Source code in src/pathlib_next/utils/glob.py

def glob(
    path: _Globable,
    *,
    dironly: bool = False,
    root_dir: _Globable | None = None,
    recursive: bool | None = False,
    include_hidden: bool = False,
    case_sensitive: bool | None = None,
    native: bool = True,
    on_error: "_ty.Callable[[OSError], None] | None" = None,
    bound_loops: bool = False,
) -> _ty.Iterable[_Globable]:
    """Return an iterator which yields the paths matching a pathname pattern.

    `path` is the pattern itself, as a path (e.g. `UriPath("file:/x/**/*.py")`).
    The pattern may contain simple shell-style wildcards a la fnmatch. Like
    the stdlib `glob` module, and unlike `Path.glob()`, hidden entries (names
    starting with a dot) are not matched by wildcards and not descended into
    by "**" unless `include_hidden` is true.

    If recursive is true, the pattern '**' will match any files and
    zero or more directories and subdirectories. `recursive=None` decides
    from the pattern itself, as `Path.glob()` does: a "**" component enables
    it. `native=` follows the running interpreter's rules, or applies one
    rule on every version -- see `parse_pattern()`.
    """
    segments = list(path.segments)
    if len(segments) > 1 and segments[-1] == "":
        segments.pop()
        if native and not _TRAILING_SLASH_SELECTS_DIRS:
            pass  # pathlib ignores a trailing separator before 3.11
        else:
            dironly = True
    if recursive is None:
        recursive = RECURSIVE in segments
    if native and not _PARTIAL_DOUBLESTAR_ALLOWED:
        for segment in segments:
            if RECURSIVE in segment and segment != RECURSIVE:
                raise ValueError(
                    "Invalid pattern: '**' can only be an entire path component"
                )
    first_wildcard = next(
        (i for i, seg in enumerate(segments) if WILDCARD_PATTERN.search(seg)),
        max(len(segments) - 1, 0),
    )
    base = path.with_segments(*segments[:first_wildcard])
    if root_dir is not None:
        base = root_dir / base
    parts = [part for part in segments[first_wildcard:] if part not in ("", ".")]
    if not parts:
        if base.is_dir() if dironly else base.exists():
            yield base
        return
    yield from select(
        base,
        parts,
        dironly=dironly,
        recursive=recursive,
        include_hidden=include_hidden,
        case_sensitive=case_sensitive,
        native=native,
        on_error=on_error,
        bound_loops=bound_loops,
    )

`parse_pattern(pattern, *, native=True)`

Split a relative glob pattern (a /-separated string or a Pathname) into its components, and report whether it ended with a separator (directories only).

Raises ValueError for an empty pattern and NonRelativePatternError for an absolute one, as pathlib.Path.glob() does. Empty and "." components are dropped.

native=True (the default) follows the running interpreter on the two rules pathlib changed mid-series: a trailing "/" is ignored before 3.11 (it selects directories only from 3.11), and a component that merely CONTAINS "" ("a") raises ValueError before 3.13. native=False applies one rule on every version instead -- a trailing "/" always selects directories only, "a**" is always a plain wildcard -- so a pattern gives the same answer on every interpreter and every backend.

Source code in src/pathlib_next/utils/glob.py

def parse_pattern(
    pattern: "str | _ty.Any", *, native: bool = True
) -> _ty.Tuple[_ty.List[str], bool]:
    """Split a relative glob `pattern` (a `/`-separated string or a
    `Pathname`) into its components, and report whether it ended with a
    separator (directories only).

    Raises `ValueError` for an empty pattern and `NonRelativePatternError`
    for an absolute one, as `pathlib.Path.glob()` does. Empty and "."
    components are dropped.

    `native=True` (the default) follows the running interpreter on the two
    rules pathlib changed mid-series: a trailing "/" is ignored before 3.11
    (it selects directories only from 3.11), and a component that merely
    CONTAINS "**" ("a**") raises `ValueError` before 3.13. `native=False`
    applies one rule on every version instead -- a trailing "/" always
    selects directories only, "a**" is always a plain wildcard -- so a
    pattern gives the same answer on every interpreter and every backend.
    """
    if isinstance(pattern, str):
        rooted = pattern.startswith("/")
        segments = pattern.split("/")
    else:
        segments = list(pattern.segments)
        rooted = bool(getattr(pattern, "anchor", "")) or bool(
            getattr(pattern, "source", None)
        )
    if rooted:
        raise NonRelativePatternError("Non-relative patterns are unsupported")
    parts = [part for part in segments if part not in ("", ".")]
    if not parts:
        raise ValueError(f"Unacceptable pattern: {str(pattern)!r}")
    if native and not _PARTIAL_DOUBLESTAR_ALLOWED:
        for part in parts:
            if RECURSIVE in part and part != RECURSIVE:
                # pathlib's own message, so an `except ValueError` that reads
                # it sees the same text it does on this interpreter.
                raise ValueError(
                    "Invalid pattern: '**' can only be an entire path component"
                )
    trailing_sep = bool(segments) and segments[-1] == ""
    if native and not _TRAILING_SLASH_SELECTS_DIRS:
        trailing_sep = False
    return parts, trailing_sep

`select(base, parts, *, dironly=False, recursive=True, include_hidden=True, case_sensitive=None, native=True, on_error=None, bound_loops=False)`

Yield the paths under base matching the pattern components parts (see parse_pattern()); the engine behind Path.glob().

Directory listings go through _scandir(); an OSError from listing a missing or non-directory path selects nothing. "" (when recursive) decides recursion from the listing's non-following stat, so it never descends into a directory symlink and always terminates. A trailing "" also selects files on Python 3.13+ and directories only before, which is the third rule native=False pins: it then selects files on every version (3.13's rule, the one current pathlib applies).

Source code in src/pathlib_next/utils/glob.py

def select(
    base: _Globable,
    parts: _ty.Sequence[str],
    *,
    dironly: bool = False,
    recursive: bool = True,
    include_hidden: bool = True,
    case_sensitive: bool | None = None,
    native: bool = True,
    on_error: "_ty.Callable[[OSError], None] | None" = None,
    bound_loops: bool = False,
) -> _ty.Iterator[_Globable]:
    """Yield the paths under `base` matching the pattern components `parts`
    (see `parse_pattern()`); the engine behind `Path.glob()`.

    Directory listings go through `_scandir()`; an `OSError` from listing a
    missing or non-directory path selects nothing. "**" (when `recursive`)
    decides recursion from the listing's non-following stat, so it never
    descends into a directory symlink and always terminates. A trailing "**"
    also selects files on Python 3.13+ and directories only before, which is
    the third rule `native=False` pins: it then selects files on every
    version (3.13's rule, the one current pathlib applies).
    """
    default_case = getattr(base, "_is_case_sensitive", True)
    if case_sensitive is None:
        case_sensitive = default_case
    if recursive:
        parts = _collapse_recursive(parts)
    steps: _ty.List[_ty.Tuple[str, _ty.Any]] = []
    for part in parts:
        if recursive and part == RECURSIVE:
            steps.append((part, None))
        elif WILDCARD_PATTERN.search(part) or case_sensitive != default_case:
            steps.append((part, compile_pattern(part, case_sensitive)))
        else:
            steps.append((part, False))
    opts = _Options(
        dironly,
        include_hidden,
        _DOUBLESTAR_SELECTS_FILES if native else True,
        on_error,
        bound_loops,
    )
    selected = _select(base, steps, 0, opts, None)
    if sum(1 for _, kind in steps if kind is None) < 2:
        yield from selected
        return
    # Two separate "**" can reach one path along several splits.
    seen = set()
    for path in selected:
        if path not in seen:
            seen.add(path)
            yield path

`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-handle 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. An st_mtime of 0 on either side means "unknown" and never matches. 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 -- decided before an existing target entry is touched. Replacing an existing entry creates the new link under a temporary sibling name first, so a runtime refusal also leaves the entry in place. A target link that already has the same raw target string (and, on a Windows target, the same file/directory link kind) is left alone. A directory link is created as one (target_is_directory), which Windows requires.

remove_missing=False never deletes a non-empty target directory: when the source entry became a file or a symlink, replacing that directory raises IsADirectoryError through ignore_error (event SyncEvent.TypeMismatch) and the directory is left untouched. With remove_missing=True, or for an empty directory, it is replaced.

A changed file is written to a temporary sibling and renamed over the existing target where the target backend implements rename(), so a failed or interrupted transfer keeps the previous version. A backend without rename() is overwritten in place, with the source opened before the target is truncated. A source entry that is not a regular file, directory or symlink (FIFO, socket, device) is skipped with a SyncEvent.Skipped event and its target left untouched.

Errors: ignore_error is consulted once per error, with the paths of the entry that failed and the event that failed (SyncEvent.Compare for a failing quick check or checksum). An error it declines propagates without being offered again by enclosing directories. A dry run takes the same decisions as a real run (a directory that would be created is treated as empty) without changing anything. Every error the policy tolerates is logged at WARNING (with the traceback) on the pathlib_next.sync logger and reported to hook as SyncEvent.Error.

hook receives PathAndStat entries for every event, and the dry_run value of the call for every event, including the structural ones (SyncStart, CheckTargetChild(ren), SyncChild(ren)).

Source code in src/pathlib_next/utils/sync.py

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.

The root call is checked before anything is touched: a source that does not exist raises FileNotFoundError (only a child that vanishes mid-sync takes the remove_missing path), and a source and target of the same implementation where one contains the other raise ValueError. Both go through ignore_error; a tolerated error ends the call without changes. Child names that would not stay a single component inside target (.., a separator or drive on a Windows target) raise ValueError the same way, and symlinks found inside target are replaced, never written, listed or deleted through.

Source code in src/pathlib_next/utils/sync.py

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.

    The root call is checked before anything is touched: a `source`
    that does not exist raises `FileNotFoundError` (only a child that
    vanishes mid-sync takes the `remove_missing` path), and a `source`
    and `target` of the same implementation where one contains the
    other raise `ValueError`. Both go through `ignore_error`; a
    tolerated error ends the call without changes. Child names that
    would not stay a single component inside `target` (`..`, a
    separator or drive on a Windows target) raise `ValueError` the
    same way, and symlinks found inside `target` are replaced, never
    written, listed or deleted through.
    """
    _ignore_error = (
        self.ignore_error
        if ignore_error is None
        else _utils.as_error_handler(ignore_error)
    )
    return self._sync(source, target, dry_run, _ignore_error, True)

`SyncEvent`

Bases: Enum

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

`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.