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,)> collections.abc.Callable[[pathlib.Path], str]:

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) and not thread-safe; 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.

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,)> collections.abc.Iterator[posix.DirEntry[str]]:

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

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.

Classes

MaxNetworkDriveRetriesExceededError

class MaxNetworkDriveRetriesExceededError(*args, **kwargs):

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