qq_lib.core.logical_paths

 1# Released under MIT License.
 2# Copyright (c) 2025-2026 Ladislav Bartos and Robert Vacha Lab
 3
 4"""
 5Symlink-preserving path resolution for shared network filesystems.
 6
 7On network storage clusters, the same filesystem is often mounted at different
 8physical paths on different machines, but is also accessible via a stable
 9symlinked path that is consistent across all machines (e.g.
10/storage/brno14-ceitec/home/user). Python's Path.resolve() and Path.cwd()
11query the kernel, which returns the physical path after resolving all symlinks.
12This makes resolved paths machine-specific and therefore non-portable when
13shared between machines (e.g. passed in a job submission).
14
15This module solves the problem by maintaining the logical current working
16directory in a module-level variable that is kept in sync with every os.chdir()
17call. os.chdir() is patched at import time to intercept all directory changes -
18including those made by third-party libraries - and update the logical cwd
19accordingly. The logical cwd is initialized from $PWD at import time, which is
20the only moment the shell-maintained logical path is reliably available.
21"""
22
23import os
24from pathlib import Path
25
26from qq_lib.core.logger import get_logger
27
28logger = get_logger(__name__)
29
30# initialized from $PWD at import time, while the shell-maintained logical path
31# is still valid; after this point, _chdir keeps it in sync
32_logical_cwd = Path(os.environ.get("PWD", Path.cwd()))
33
34_real_chdir = os.chdir
35
36
37def _chdir(path: str | Path) -> None:
38    """
39    Replacement for os.chdir that keeps _logical_cwd in sync.
40
41    The shell maintains $PWD as the logical current working directory, updating
42    it on every cd. Python's os.chdir does not do this, causing the logical cwd
43    to be unknowable after the first directory change. This replacement
44    replicates the shell's behaviour by computing the new logical path
45    lexically - without following symlinks - and storing it in _logical_cwd
46    before delegating to the real os.chdir.
47
48    Args:
49        path (str | Path): The directory to change to. May be relative or absolute.
50    """
51    global _logical_cwd
52    path = Path(path)
53    if not path.is_absolute():
54        path = _logical_cwd / path
55    _logical_cwd = Path(os.path.normpath(path))
56    _real_chdir(path)
57    logger.debug(
58        f"Hijacking `os.chdir` to update `_logical_cwd`. Current `_logical_cwd` is '{_logical_cwd}'."
59    )
60
61
62# replacing the standard os.chdir with the patched version
63os.chdir = _chdir  # type: ignore
64
65
66def logical_resolve(path: Path, base: Path | None = None) -> Path:
67    """
68    Resolve a path to a logical absolute path without expanding symlinks.
69
70    `Path.resolve()` and `Path.cwd()` both return the physical absolute
71    path by querying the kernel, which follows all symlinks. On shared network
72    filesystems this produces machine-specific paths that cannot be shared
73    across machines. This function instead produces the logical absolute path,
74    which remains consistent across all machines that mount the same filesystem
75    via the same symlinked prefix.
76
77    Relative paths are anchored to `base` (or the logical current working
78    directory if `base` is not given), then `.` and `..` components are
79    collapsed lexically using `os.path.normpath`. No filesystem access is
80    performed, so symlinks are never followed.
81
82    Args:
83        path (Path): The path to resolve. May be relative or absolute.
84        base (Path | None): The base directory to anchor relative paths to. Must be a
85            logical absolute path. Defaults to the logical current working
86            directory maintained by the os.chdir patch.
87
88    Returns:
89        Path: A logical absolute path with `.` and `..` components resolved
90        but symlinks left intact.
91    """
92    if not path.is_absolute():
93        if base is None:
94            base = _logical_cwd
95        path = base / path
96    return Path(os.path.normpath(path))