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.
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 optionalstat_resultobject for the file, as returned byos.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 optionalstat_resultobject for the file, as returned byos.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 optionalstat_resultobject for the file, as returned byos.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 optionalstat_resultobject for the path, as returned byos.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 tonormalize_pathfor each parent directory.
Returns
A callable mapping a walked Path to its normalized string spelling.
Raises
OSError: From the returned callable, only whenstrict=Trueand a parent directory could not be resolved.FileNotFoundErrordistinguishes "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 underlyingOSError(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 whenstrict=Trueand 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.