File Operations

Functions for working with files and directories.

---

ArchiveType

Bases: Enum

Enumeration of supported archive formats.

Maps to formats recognized by :func:shutil.make_archive and :func:shutil.unpack_archive.

archive_format property

archive_format

Return the shutil-compatible archive format string.

from_path classmethod

from_path(path)

Infer the archive type from a file path.

Parameters:

Name	Type	Description	Default
path	Path	Path to an existing archive file.	required

Returns:

Type	Description
ArchiveType	The inferred ArchiveType.

Raises:

Type	Description
ValueError	If the path does not exist, is a directory, or is not a recognized archive.

Source code in src/aibs_informatics_core/utils/file_operations.py

@classmethod
def from_path(cls, path: Path) -> "ArchiveType":
    """Infer the archive type from a file path.

    Args:
        path: Path to an existing archive file.

    Returns:
        The inferred ``ArchiveType``.

    Raises:
        ValueError: If the path does not exist, is a directory, or is not a recognized archive.
    """
    if not path.exists() or path.is_dir():
        raise ValueError(f"Path {path} does not exist or is a directory")
    if tarfile.is_tarfile(path):
        compression_type = cls._get_compression_type(path)
        if compression_type == "gz":
            return ArchiveType.TAR_GZ
        elif compression_type == "bz":
            return ArchiveType.TAR_BZ
        elif compression_type == "xz":
            return ArchiveType.TAR_XZ
        else:
            return ArchiveType.TAR
    elif zipfile.is_zipfile(path):
        return ArchiveType.ZIP
    else:
        raise ValueError("Cannot infer type")

is_archive classmethod

is_archive(path)

Check if a path is any supported archive type.

Parameters:

Name	Type	Description	Default
path	Path	Path to check.	required

Returns:

Type	Description
bool	True if path is a recognized archive.

Source code in src/aibs_informatics_core/utils/file_operations.py

@classmethod
def is_archive(cls, path: Path) -> bool:
    """Check if a path is any supported archive type.

    Args:
        path: Path to check.

    Returns:
        True if ``path`` is a recognized archive.
    """
    try:
        cls.from_path(path)
    except Exception:
        return False
    else:
        return True

is_archive_type

is_archive_type(path)

Check if a path matches this archive type.

Parameters:

Name	Type	Description	Default
path	Path	Path to check.	required

Returns:

Type	Description
bool	True if path is an archive of this type.

Source code in src/aibs_informatics_core/utils/file_operations.py

def is_archive_type(self, path: Path) -> bool:
    """Check if a path matches this archive type.

    Args:
        path: Path to check.

    Returns:
        True if ``path`` is an archive of this type.
    """
    try:
        return self == self.from_path(path)
    except Exception:
        return False

CannotAcquirePathLockError

Bases: Exception

Raised when a path lock cannot be acquired.

PathLock dataclass

PathLock(path, lock_root=None, raise_if_locked=False)

A context manager for acquiring and releasing locks on a file or directory path.

If lock_root is provided, a lock file will be created in that directory with the name of the hash of the path. If lock_root is not provided, a lock file with the same name as the path and a .lock extension will be created.

Providing an explicit lock root is useful if you dont want processes to read the lock file from the same directory as the file being locked.

Attributes:

Name	Type	Description
path	Union[str, Path]	The path to the file.
lock_root	Optional[Union[str, Path]]	The root directory for lock files. If provided, a lock file will be created in this directory with the name of the hash of the path. Otherwise, a lock file with the same name as the path and a .lock extension will be created. Defaults to None.

acquire

acquire()

Acquire the file lock.

Raises:

Type	Description
CannotAcquirePathLockError	If the lock cannot be acquired.

Source code in src/aibs_informatics_core/utils/file_operations.py

def acquire(self):
    """Acquire the file lock.

    Raises:
        CannotAcquirePathLockError: If the lock cannot be acquired.
    """
    logger.info("Acquiring lock...")
    try:
        self._lock_path.parent.mkdir(parents=True, exist_ok=True)
        self._lock_file = open(self._lock_path, "w")
        op = fcntl.LOCK_EX
        if self.raise_if_locked:
            op |= fcntl.LOCK_NB
        fcntl.flock(self._lock_file, op)
        self._lock_file.write(f"{datetime.now().timestamp()}")
        logger.info("Lock acquired!")
    except Exception as e:
        msg = f"Could not acquire lock! Reason: {e}"
        logger.error(msg)
        raise CannotAcquirePathLockError(msg) from e

release

release()

Release the file lock and remove the lock file.

Source code in src/aibs_informatics_core/utils/file_operations.py

def release(self):
    """Release the file lock and remove the lock file."""
    logger.info("Releasing lock...")

    if self._lock_file and not self._lock_file.closed:
        try:
            fcntl.flock(self._lock_file, fcntl.LOCK_UN)
            self._lock_file.close()
        except Exception as e:
            logger.warning(f"Lock file doesn't exist. Skipping fcntl.flock and close: {e}")
    else:
        logger.warning("Strange! lock file already closed. not calling fcntl.flock")
    logger.info("Removing lock file")
    remove_path(self._lock_path)

    logger.info("Lock released!")

copy_path

copy_path(source_path, destination_path, exists_ok=False)

Copies path from source to destination

Parameters:

Name	Type	Description	Default
source_path	Path	source path	required
destination_path	Path	destination path	required
exists_ok	bool	if true, overwrites destination. Defaults to False.	False

Source code in src/aibs_informatics_core/utils/file_operations.py

def copy_path(source_path: Path, destination_path: Path, exists_ok: bool = False):
    """Copies path from source to destination

    Args:
        source_path (Path): source path
        destination_path (Path): destination path
        exists_ok (bool, optional): if true, overwrites destination. Defaults to False.

    """
    if source_path.is_file():
        if destination_path.exists():
            if not exists_ok:
                raise ValueError(f"Cannot copy path to {destination_path}. destination exists!")
            remove_path(destination_path)
        shutil.copy(source_path, destination_path)
    else:
        shutil.copytree(source_path, destination_path, dirs_exist_ok=exists_ok)

extract_archive

extract_archive(source_path, destination_path=None)

Untar/unzip data batch into a dedicate folder Example: batch_of_samples.tar.gz -> batch_of_samples

Parameters:

Name	Type	Description	Default
source_path	Path	archived batch data	required
destination_path	Optional[Path]	Optional destination path for extracted data. If none, then source path name used with extension removed.	None

Raises:

Type	Description
RuntimeError	If extraction fails

Returns:

Type	Description
Path	path to the untarred data

Source code in src/aibs_informatics_core/utils/file_operations.py

def extract_archive(source_path: Path, destination_path: Path | None = None) -> Path:
    """Untar/unzip data batch into a dedicate folder
    Example: batch_of_samples.tar.gz -> batch_of_samples

    Args:
        source_path (Path): archived batch data
        destination_path (Optional[Path]): Optional destination path for extracted data.
            If none, then source path name used with extension removed.

    Raises:
        RuntimeError: If extraction fails

    Returns:
        path to the untarred data
    """
    archive_type = ArchiveType.from_path(source_path)

    extracted_path = destination_path or Path(tempfile.mkdtemp())

    shutil.unpack_archive(
        filename=str(source_path),
        extract_dir=str(extracted_path),
        format=archive_type.archive_format,
    )
    return extracted_path

find_filesystem_boundary

find_filesystem_boundary(starting_path)

Given some starting Path, determine the nearest filesystem boundary (mount point). If no mount is found, then this function will return the first parent directory PRIOR to the filesystem anchor.

find_filesystem_boundary(Path("/allen/scratch/aibstemp")) PosixPath('/allen/scratch')

find_filesystem_boundary(Path("/tmp/random_file.txt")) PosixPath('/tmp')

Parameters:

Name	Type	Description	Default
starting_path	Path	The starting Path	required

Raises:

Type	Description
RuntimeError	If the provided starting_path cannot resolve to a real existing path

Returns:

Type	Description
Path	The path of the nearest filesystem boundary OR the first parent directory prior to the filesystem anchor (example anchors: "/", "c:\")

Source code in src/aibs_informatics_core/utils/file_operations.py

def find_filesystem_boundary(starting_path: Path) -> Path:
    """Given some starting Path, determine the nearest filesystem boundary (mount point).
    If no mount is found, then this function will return the first parent directory PRIOR
    to the filesystem anchor.

    >>> find_filesystem_boundary(Path("/allen/scratch/aibstemp"))
    PosixPath('/allen/scratch')

    >>> find_filesystem_boundary(Path("/tmp/random_file.txt"))
    PosixPath('/tmp')

    Args:
        starting_path (Path): The starting Path

    Raises:
        RuntimeError: If the provided starting_path cannot resolve to a real existing path

    Returns:
        The path of the nearest filesystem boundary OR the first parent directory prior
            to the filesystem anchor (example anchors: "/", "c:\\")
    """
    current_path = starting_path.resolve()
    while current_path.parent != Path(current_path.anchor):
        if current_path.is_mount():
            break
        current_path = current_path.parent

    if current_path.exists():
        return current_path
    else:
        raise OSError(f"Could not find a real filesystem boundary for: {str(starting_path)}")

find_paths

find_paths(
    root,
    include_dirs=True,
    include_files=True,
    includes=None,
    excludes=None,
)

Find paths that match criteria

Parameters:

Name	Type	Description	Default
root	Union[str, Path]	root path	required
include_dirs	bool	whether to include directories. Defaults to True.	True
include_files	bool	whether to include files. Defaults to True.	True
includes	Sequence[str]	list of regex patterns to include. Defaults to all.	None
excludes	Sequence[str]	list of regex patterns to exclude. Defaults to None.	None

Returns:

Type	Description
list[str]	list of paths matching criteria

Source code in src/aibs_informatics_core/utils/file_operations.py

def find_paths(
    root: str | Path,
    include_dirs: bool = True,
    include_files: bool = True,
    includes: Sequence[Pattern | str] | None = None,
    excludes: Sequence[Pattern | str] | None = None,
) -> list[str]:
    """Find paths that match criteria

    Args:
        root (Union[str, Path]): root path
        include_dirs (bool, optional): whether to include directories. Defaults to True.
        include_files (bool, optional): whether to include files. Defaults to True.

        includes (Sequence[str], optional): list of regex patterns to include. Defaults to all.
        excludes (Sequence[str], optional): list of regex patterns to exclude. Defaults to None.

    Returns:
        list of paths matching criteria
    """

    paths_to_return = []
    paths = find_all_paths(root, include_dirs=include_dirs, include_files=include_files)

    if not includes and not excludes:
        return paths

    include_patterns = [re.compile(include) for include in includes or [r".*"]]
    exclude_patterns = [re.compile(exclude) for exclude in excludes or []]

    for path in paths:
        target_path = path  # if check_absolute_path else strip_path_root(root=root, path=path)
        # First check exclude patterns
        for exclude_pattern in exclude_patterns:
            if exclude_pattern.fullmatch(target_path):
                break
        else:
            # Now check include patterns
            for include_pattern in include_patterns:
                if include_pattern.fullmatch(target_path):
                    paths_to_return.append(path)
                    break
    return paths_to_return

get_path_size_bytes

get_path_size_bytes(path)

Calculate the total size in bytes of all files under a path.

Handles FileNotFoundError and stale NFS file handles gracefully.

Parameters:

Name	Type	Description	Default
path	Path	A file or directory path.	required

Returns:

Type	Description
int	The total size in bytes.

Source code in src/aibs_informatics_core/utils/file_operations.py

def get_path_size_bytes(path: Path) -> int:
    """Calculate the total size in bytes of all files under a path.

    Handles ``FileNotFoundError`` and stale NFS file handles gracefully.

    Args:
        path: A file or directory path.

    Returns:
        The total size in bytes.
    """
    size_bytes = 0
    file_paths = deque(find_all_paths(path, include_dirs=False, include_files=True))
    while file_paths:
        file_path = file_paths.popleft()
        path = Path(file_path)
        try:
            size_bytes += path.stat().st_size
        except FileNotFoundError as e:
            logger.warning(f"File at {path} does not exist anymore. Reason: {e}")
            continue
        except OSError as ose:
            if ose.errno == errno.ESTALE:
                logger.warning(f"{ose} raised for {file_path}.")
                try:
                    if Path(str(path)).exists():
                        logger.warning(f"Adding {path} to end of list to check later.")
                        file_paths.append(file_path)
                except (FileNotFoundError, OSError) as fall_back_e:
                    logger.warning(f"Failed to check if path {path} exists: {fall_back_e}")
                    continue
            else:
                logger.error(f"Unexpected error raised for {path}. Reason: {ose}")
                raise ose
    return size_bytes

get_path_with_root

get_path_with_root(path, root)

Ensure a path is rooted under the given directory.

If path is already relative to root, it is returned unchanged. Otherwise, the path is made relative and joined with root.

Parameters:

Name	Type	Description	Default
path	str | Path	The path to adjust.	required
root	str | Path	The root directory.	required

Returns:

Type	Description
str	The normalized path string under root.

Source code in src/aibs_informatics_core/utils/file_operations.py

def get_path_with_root(path: str | Path, root: str | Path) -> str:
    """Ensure a path is rooted under the given directory.

    If ``path`` is already relative to ``root``, it is returned unchanged.
    Otherwise, the path is made relative and joined with ``root``.

    Args:
        path: The path to adjust.
        root: The root directory.

    Returns:
        The normalized path string under ``root``.
    """
    orig_path = path
    root = Path(root)
    path = Path(path)
    if path.is_relative_to(root):
        return str(orig_path)
    rel_path = path.relative_to("/") if path.is_absolute() else path
    path_with_root = os.path.normpath(root / rel_path)
    return path_with_root

make_archive

make_archive(
    source_path,
    destination_path=None,
    archive_type=ArchiveType.TAR_GZ,
)

tar/zip data batch from a folder Example: batch_of_samples -> batch_of_samples.tar.gz

Parameters:

Name	Type	Description	Default
source_path	Path	folder of data to archive	required
destination_path	Optional[Path]	Optional destination path for archived file. If none, then tmp file is created and used	None

Raises:

Type	Description
RuntimeError	If archiving operation fails

Returns:

Type	Description
Path	path to the untarred data

Source code in src/aibs_informatics_core/utils/file_operations.py

def make_archive(
    source_path: Path,
    destination_path: Path | None = None,
    archive_type: ArchiveType | ArchiveFormat = ArchiveType.TAR_GZ,
) -> Path:
    """tar/zip data batch from a folder
    Example: batch_of_samples -> batch_of_samples.tar.gz

    Args:
        source_path (Path): folder of data to archive
        destination_path (Optional[Path]): Optional destination path for archived file.
            If none, then tmp file is created and used

    Raises:
        RuntimeError: If archiving operation fails

    Returns:
        path to the untarred data
    """

    archive_path = destination_path or Path(tempfile.mktemp())
    if not isinstance(archive_type, ArchiveType):
        archive_type = ArchiveType(archive_type)

    try:
        actual_archive_path = Path(
            shutil.make_archive(
                base_name=str(archive_path.with_suffix("")),
                root_dir=str(source_path),
                base_dir=None,
                format=archive_type.archive_format,
            )
        )

        # shutil.make_archive takes base name and appends the appropriate suffix.
        # This means we can't say "write at X". We must move results, but only if different.
        if actual_archive_path != archive_path:
            move_path(actual_archive_path, archive_path)
        return archive_path
    except Exception as e:
        raise ValueError(f"Error extracting file {source_path}. [{e}]") from e

move_path

move_path(source_path, destination_path, exists_ok=False)

Alias to simple mv command from one path to another

Parameters:

Name	Type	Description	Default
source_path	Path	source path	required
destination_path	Path	destination path	required
exists_ok	bool	if true, overwrites destination. Defaults to False.	False

Source code in src/aibs_informatics_core/utils/file_operations.py

def move_path(source_path: Path, destination_path: Path, exists_ok: bool = False):
    """Alias to simple mv command from one path to another

    Args:
        source_path (Path): source path
        destination_path (Path): destination path
        exists_ok (bool, optional): if true, overwrites destination. Defaults to False.

    """
    if destination_path.exists():
        if not exists_ok:
            raise ValueError(f"Cannot move path to {destination_path}. destination exists!")
        remove_path(destination_path)

    shutil.move(str(source_path), destination_path)

remove_path

remove_path(path, ignore_errors=True)

Remove a file or directory at the given path.

Parameters:

Name	Type	Description	Default
path	Path	The path to remove.	required
ignore_errors	bool	If True, errors are logged but not raised. Defaults to True.	True

Source code in src/aibs_informatics_core/utils/file_operations.py

def remove_path(path: Path, ignore_errors: bool = True):
    """Remove a file or directory at the given path.

    Args:
        path: The path to remove.
        ignore_errors: If True, errors are logged but not raised. Defaults to True.
    """
    try:
        if path.exists():
            if path.is_dir():
                shutil.rmtree(path, ignore_errors=ignore_errors)
            else:
                os.remove(path)
    except FileNotFoundError as e:
        # Ignore errors if requested
        if not ignore_errors:
            raise e
    except OSError as ose:
        # Ignore errors if requested
        if not ignore_errors:
            raise ose
        else:
            logger.warning(f"Failed to remove path {path}. Reason: {ose}")
            return

strip_path_root

strip_path_root(path, root=None)

Strip the root from the path if path is absolute

Parameters:

Name	Type	Description	Default
path	Union[str, Path]	path to strip root from	required
root	Optional[Union[str, Path]]	optionally specify root. If no root specified, uses "/".	None

Returns:

Type	Description
str	a relative path

Source code in src/aibs_informatics_core/utils/file_operations.py

def strip_path_root(path: str | Path, root: str | Path | None = None) -> str:
    """Strip the root from the path if path is absolute

    Args:
        path (Union[str, Path]): path to strip root from
        root (Optional[Union[str, Path]], optional): optionally specify root.
            If no root specified, uses "/".

    Returns:
        a relative path
    """
    root = Path(root) if root is not None else Path("/")
    path = Path(path)
    rel_path = path.relative_to(root) if path.is_absolute() else path
    return os.path.normpath(rel_path)