Skip to main content

fs_utils

Utility functions to interact with the filesystem.

Module​

Functions​

get_file_creation_date​

def get_file_creation_date(    path: str | os.PathLike[str], stat: os.stat_result | None = None,) ‑> datetime.date:

Get the creation date of a file with consideration for different OSs.

If the stat object is provided, then the information will be extracted from this in preference to getting a new one from the filesystem. Note that there is no checking that the stat object is up-to-date or even corresponds to the same file as path, so care should be taken to pass through the correct object.

caution

It is not possible to get the creation date of a file on Linux. This method will return the last modification date instead. This will impact filtering of files by date.

Arguments

  • path: The path to the file.
  • stat: An optional stat_result object for the file, as returned by os.stat(path). This can be used to avoid making a new filesystem query.

Returns The creation date of the file.

get_file_last_modification_date​

def get_file_last_modification_date(    path: str | os.PathLike[str], stat: os.stat_result | None = None,) ‑> datetime.date:

Get the last modification date of a file.

If the stat object is provided, then the information will be extracted from this in preference to getting a new one from the filesystem. Note that there is no checking that the stat object is up-to-date or even corresponds to the same file as path, so care should be taken to pass through the correct object.

Arguments

  • path: The path to the file.
  • stat: An optional stat_result object for the file, as returned by os.stat(path). This can be used to avoid making a new filesystem query.

Returns The last modification date of the file.

get_file_size​

def get_file_size(    path: str | os.PathLike[str], stat: os.stat_result | None = None,) ‑> int:

Get the size, in bytes, of a file on the filesystem.

If the stat object is provided, then the information will be extracted from this in preference to getting a new one from the filesystem. Note that there is no checking that the stat object is up-to-date or even corresponds to the same file as path, so care should be taken to pass through the correct object.

Arguments

  • path: The path to the file.
  • stat: An optional stat_result object for the file, as returned by os.stat(path). This can be used to avoid making a new filesystem query.

Returns The size of the file, in bytes.

is_file​

def is_file(    path: str | os.PathLike[str] | os.DirEntry[str], stat: os.stat_result | None = None,) ‑> bool:

Determine if a path is a file or not.

If the stat object is provided, then the information will be extracted from this in preference to getting a new one from the filesystem. Note that there is no checking that the stat object is up-to-date or even corresponds to the same file as path, so care should be taken to pass through the correct object.

Arguments

  • path: The path to check. Can also be an os.DirEntry as from scandir() or scantree().
  • stat: An optional stat_result object for the path, as returned by os.stat(path). This can be used to avoid making a new filesystem query.

Returns The size of the file, in bytes.

memoized_path_normalizer​

def memoized_path_normalizer(    *, strict: bool = False,) ‑> MemoizedPathNormalizer:

Return a Path -> normalized str function that memoizes per parent directory.

normalize_path costs a filesystem round-trip, which is not affordable per file on a network share: a walk of a million files would pay a million of them. But alias-ness (symlink, Windows junction, mapped drive) is a property of the directories on the way to a file, not of the file itself, so each parent directory is resolved once and reused for every file within it. In a typical walk that is one round-trip per directory and a dict lookup per file.

Only the parent is resolved; the file's own name is joined onto it verbatim. For a walk that is what you want — scandir yields the true on-disk name, so the case is already canonical and a symlinked file keeps its own spelling, matching what the caches hold. It does mean this is not equivalent to normalize_path on an arbitrary path: hand it a caller-supplied string whose filename differs in case, or is an 8.3 short name, and that component comes back unchanged. Use normalize_path directly for those.

Pass strict=True when the result becomes a cache key, so a directory that resolves only partially raises instead of yielding a plausible-looking mapped-drive spelling — see normalize_path's note on Windows. Because the memo is per directory, one unnoticed half-resolution would otherwise mis-key every file beneath that directory, not just one.

The returned function is stateful (it owns the memo); make one per walk rather than sharing one globally, so the memo cannot go stale against a directory tree that changed between walks. Failures are not memoized, so a directory that recovers resolves correctly on the next file. It is safe to share between threads, and a pooled walk can prime it with each directory as it lists it.

Arguments

  • strict: Propagated to normalize_path for each parent directory.

Returns A callable mapping a walked Path to its normalized string spelling.

Raises

  • OSError: From the returned callable, only when strict=True and a parent directory could not be resolved. FileNotFoundError distinguishes "the directory is gone" (its files are gone too, so skipping them is correct) from "unreachable" (nothing beneath it can be trusted).

normalize_path​

def normalize_path(path: str | os.PathLike[str], *, strict: bool = False) ‑> pathlib.Path:

Normalize a path to handle mapped drives and UNC paths equivalently.

This function resolves mapped drives (e.g., S:\patients) to their UNC equivalents (e.g., \FSC\Filestorage1\Images\patients) so that paths referring to the same location are treated as equivalent.

On Windows, Path.resolve() automatically resolves mapped drives to UNC paths. On other platforms, this function resolves symlinks and relative paths.

When network_drive_robustness is enabled and the path appears to be on a network drive, transient OSError failures from resolve() are retried with exponential backoff before falling back to an absolute path.

Why strict matters on Windows​

Path.resolve(strict=False) does not fail loudly when it cannot resolve a path. On Windows it resolves the deepest prefix it can and joins the rest of the path back as originally spelled (ntpath._getfinalpathname_nonstrict), swallowing the network errors that matter here — ERROR_BAD_NETPATH (53), ERROR_NETWORK_ACCESS_DENIED (65), ERROR_BAD_NET_NAME (67), ERROR_NOT_READY (21). A share that blinks therefore yields a mapped-drive spelling that looks like a successful result, no exception is raised, and the retry logic above never runs.

For a caller that is about to use the result as a cache key that is not an acceptable answer: the two spellings of one file would be written into different primary keys depending on network weather. Pass strict=True there. It turns the silent half-resolution into an OSError, which makes the network retry reachable, and re-raises if the path still cannot be resolved so the caller can abort rather than persist a suspect key.

Leave strict at False for paths that need not exist (an output directory being computed, a file being probed), where a best-effort absolute path is the useful answer.

Arguments

  • path: The path to normalize.
  • strict: When True, require that the path actually resolve: raise the underlying OSError (after any retries) instead of falling back to an unresolved absolute path.

Returns A normalized and absolute Path object that can be used for path comparisons and operations like relative_to().

Raises

  • OSError: Only when strict=True and the path could not be resolved. FileNotFoundError (a subclass) specifically means the path is not there, as distinct from being unreachable.

path_is_under​

def path_is_under(    path: str | os.PathLike[str], directory: str | os.PathLike[str],) ‑> bool:

Return whether path sits inside directory.

Purely lexical: no filesystem access, so this is safe to call once per row of a large inventory. Comparing Path objects (rather than string prefixes) makes it robust to redundant separators and to the trailing-separator ambiguity a startswith test has. directory is not considered to be under itself.

Neither side is resolved here — callers that need symlink/.. normalisation should resolve both sides first so the comparison is apples-to-apples. See path_relative_to_base for the resolving, symlink-aware (and much more expensive) variant.

Arguments

  • path: The candidate path.
  • directory: The directory the candidate must be within.

Returns True when path is strictly within directory.

path_relative_to_base​

def path_relative_to_base(    base: str | os.PathLike[str],    file_path: str | os.PathLike[str],    *,    max_symlink_depth: int = 1,) ‑> pathlib.Path:

Path of file_path relative to base.

(1) Resolved paths first: uses normalize_path on both and relative_to to check if they are relative (handles UNC / mapped drive equivalence).

(2) If that fails and symlinks are enabled in config (follow_symlinks_in_scantree), uses a BFS of paths under base up to max_symlink_depth (cached per base so many files under the same folder only trigger one walk): if file_path's resolved form lies under a symlink target, returns the corresponding relative path. When network_drive_robustness is enabled, the BFS uses robust scanning.

Arguments

  • base: The base (e.g. datasource root) path.
  • file_path: The file path to express relative to base.
  • max_symlink_depth: When the symlink fallback runs, how many directory levels under base to consider (1 = top level only). Default 1.

Returns A Path relative to base (e.g. "Y/foo" when base is /X and file is under /X/Y or under the resolved target of /X/Y).

Raises

  • ValueError: If file_path is not under base.

safe_append_to_file​

def safe_append_to_file(    func: Callable[[Path], R], initial_path: Path,) ‑> tuple[R, pathlib.Path]:

Handle PermissionError when appending to a file.

Execute some function that writes/appends to a file and if it's not possible to append due to a PermissionError (e.g. the user has opened the file in Windows so can't be appended to) try backup paths to either create or append to.

The supplied func should take a single Path argument and return the result, but should be able to handle the case when the file doesn't exist (i.e. writing fresh) and when a file does exist but needs to be appended to.

Arguments

  • func: Function to execute, that takes in the destination file path and will append to an existing file or write to a new file.
  • initial_path: The desired destination file path.

Returns A tuple of the result of the function and the actual path finally appended/written to.

safe_write_to_file​

def safe_write_to_file(    func: Callable[[Path], R], initial_path: Path,) ‑> tuple[R, pathlib.Path]:

Handle PermissionError when writing to a file.

Execute some function that writes to a file and if it's not possible to write due to a PermissionError (e.g. the user has opened the file in Windows so can't be appended to) try to write to a new file instead.

Arguments

  • func: Function to execute, that takes in the destination file path.
  • initial_path: The desired destination file path.

Returns A tuple of the result of the function and the actual path finally written to.

scantree​

def scantree(    root: str | os.PathLike[str],    *,    follow_symlinks: bool | None = None,    workers: int = 1,    stat_files: bool = False,    on_directory: Callable[[str], None] | None = None,    on_skipped: OnSkipped | None = None,) ‑> collections.abc.Iterator[posix.DirEntry[str]]:

Recursively iterate through a folder as in scandir(), yielding file entries.

A directory or entry that cannot be read is warned about and skipped, and passed to on_skipped when given.

Arguments

  • root: Root path to iterate from.
  • follow_symlinks: If True, follow symbolic links to directories when recursing. If False, do not recurse into symlinked directories. If None (default), use config.settings.follow_symlinks_in_scantree.
  • workers: How many directories to list at once. Above 1 the walk is breadth-first and yields in an undefined order; worth it where each listing is a network round trip. Ignored in network-drive robustness mode, which walks serially.
  • stat_files: With workers above 1, stat each file on the pool so that the caller's entry.stat() is already cached. Only worth it where the caller stats every file and DirEntry does not already carry the result (it does on Windows).
  • on_directory: With workers above 1, called on the pool with each directory before its files are yielded, such as MemoizedPathNormalizer.prime. Must not raise. Not called by the serial walks.
  • on_skipped: Called with each path the walk could not read and the error, on whichever thread read it. Must not raise.

Classes​

MaxNetworkDriveRetriesExceededError​

class MaxNetworkDriveRetriesExceededError(*args, **kwargs):

Exception raised when the max number of network drive retries exceeded.

Ancestors​

MemoizedPathNormalizer​

class MemoizedPathNormalizer(*, strict: bool):

A Path -> normalized str function that memoizes per parent directory.

Built by memoized_path_normalizer, which documents the behaviour. Safe to call and to prime from several threads.

Arguments

  • strict: Propagated to normalize_path for each parent directory.

Methods​

prime​

def prime(self, directory: str | os.PathLike[str]) ‑> None:

Resolve directory ahead of its files, so their lookups hit the memo.

For a pooled walk, which calls this on the thread that lists the directory: the round trip then overlaps others instead of happening on the consumer, one directory at a time. A failure stores nothing, so the consumer's own lookup resolves the directory again and raises as it would have.

Arguments

  • directory: A directory whose files will be looked up next, spelled as the walk spells their parent.

RootProbe​

class RootProbe(    root: str | os.PathLike[str],    *,    ttl_seconds: float = 10.0,    clock: Callable[[], float] = <built-in function monotonic>,):

Whether a datasource root is reachable, without re-probing a dead share.

The arbiter of whether a per-file or per-directory failure is transient: a failure while the root is unreachable says nothing about the file, so it is left for a later pass; with the root reachable, the failure is the file's own. Consulted only on failure paths.

Only an unreachable answer is reused, for the TTL: a probe of a dead share can block until the SMB timeout, while a reachable one is a quick stat. A reachable answer is never reused by a later failure, since the share may have dropped in between and the file would be blamed for it. The probe is single-flight: callers that arrive while one is running wait for it and share its answer, whichever it is.

Arguments

  • root: The datasource root.
  • ttl_seconds: How long an unreachable answer is reused.
  • clock: Monotonic clock, injectable for tests.

Variables​

  • root : str - The probed root.

Methods​

reachable​

def reachable(self) ‑> bool:

Whether the root answers a stat now, or was unreachable just now.