File System

File system abstractions for data synchronization.

---

BaseFileSystem dataclass

BaseFileSystem()

kept_objects property

kept_objects

Number of objects retained by the last refresh.

partition

partition(
    size_bytes_limit=None,
    object_count_limit=None,
    raise_error_if_criteria_not_met=False,
)

Partitions the root tree folder structure into a list of nodes.

Partitioning is guided by constraints by size and object count.

Parameters:

Name	Type	Description	Default
size_bytes_limit	int | None	If specified, partitions must be less than the specified value.	None
object_count_limit	int | None	If specified, partitions must contain fewer objects than the specified value.	None
raise_error_if_criteria_not_met	bool	If True, raises error if nodes cannot meet criteria. In actuality, this is more relevant for size limitations where an object size is greater than the size limit.	False

Raises:

Type	Description
ValueError	Thrown if raise_error_if_criteria_not_met is true and criteria not met.

Returns:

Type	Description
list[Node]	List of nodes representing the partition.

Source code in src/aibs_informatics_aws_utils/data_sync/file_system.py

def partition(
    self,
    size_bytes_limit: int | None = None,
    object_count_limit: int | None = None,
    raise_error_if_criteria_not_met: bool = False,
) -> list[Node]:
    """Partitions the root tree folder structure into a list of nodes.

    Partitioning is guided by constraints by size and object count.

    Args:
        size_bytes_limit: If specified, partitions must be less than the specified value.
        object_count_limit: If specified, partitions must contain fewer objects than
            the specified value.
        raise_error_if_criteria_not_met: If True, raises error if nodes cannot meet
            criteria. In actuality, this is more relevant for size limitations where
            an object size is greater than the size limit.

    Raises:
        ValueError: Thrown if raise_error_if_criteria_not_met is true and criteria not met.

    Returns:
        List of nodes representing the partition.
    """
    unchecked_nodes = {self.node}
    size_bytes_exceeding_obj_nodes = []

    partitioned_nodes: list[Node] = []
    logger.info(
        f"Partitioning nodes with size_bytes_limit={size_bytes_limit} "
        f"and object_count_limit={object_count_limit}"
    )

    while unchecked_nodes:
        unchecked_node = unchecked_nodes.pop()
        if (size_bytes_limit and unchecked_node.size_bytes > size_bytes_limit) or (
            object_count_limit and unchecked_node.object_count > object_count_limit
        ):
            if unchecked_node.has_children():
                unchecked_nodes.update(unchecked_node.children.values())
            else:
                size_bytes_exceeding_obj_nodes.append(unchecked_node)
        else:
            partitioned_nodes.append(unchecked_node)

    if size_bytes_exceeding_obj_nodes:
        msg = (
            f"Found {len(size_bytes_exceeding_obj_nodes)} objects that exceed the "
            f"partition size limit {size_bytes_limit}."
        )
        if raise_error_if_criteria_not_met:
            raise ValueError(msg)
        logger.warning(msg)
        partitioned_nodes.extend(size_bytes_exceeding_obj_nodes)
    logger.info(f"Partitioned {len(partitioned_nodes)} nodes.")
    return partitioned_nodes

refresh abstractmethod

refresh(filter_config=None, filter_root=None, **kwargs)

Rebuild the tree, optionally keeping only the paths that pass filters.

Parameters:

Name	Type	Description	Default
filter_config	DataSyncFilterConfig | None	Optional include/exclude filters. When given, only matching objects contribute to the tree -- and therefore to the sizes that partition bins on.	None
filter_root	str | None	Root that patterns are matched relative to. Defaults to this file system's own root. The distributed sync workflow splits a sync into sub-requests rooted at sub-prefixes, and those must pass the original root here or patterns stop matching.	None
**kwargs		Additional arguments passed to the underlying client.	{}

Source code in src/aibs_informatics_aws_utils/data_sync/file_system.py

@abstractmethod
def refresh(
    self,
    filter_config: DataSyncFilterConfig | None = None,
    filter_root: str | None = None,
    **kwargs,
):
    """Rebuild the tree, optionally keeping only the paths that pass filters.

    Args:
        filter_config: Optional include/exclude filters. When given, only
            matching objects contribute to the tree -- and therefore to the
            sizes that `partition` bins on.
        filter_root: Root that patterns are matched relative to. Defaults to
            this file system's own root. The distributed sync workflow splits
            a sync into sub-requests rooted at sub-prefixes, and those must
            pass the *original* root here or patterns stop matching.
        **kwargs: Additional arguments passed to the underlying client.
    """
    raise NotImplementedError()

resolve_filter_root abstractmethod

resolve_filter_root(filter_root)

Resolve the root that filter patterns are matched relative to.

Parameters:

Name	Type	Description	Default
filter_root	str | None	Explicit root, or None to use this file system's own root.	required

Returns:

Type	Description
str	The root to relativize paths against.

Source code in src/aibs_informatics_aws_utils/data_sync/file_system.py

@abstractmethod
def resolve_filter_root(self, filter_root: str | None) -> str:
    """Resolve the root that filter patterns are matched relative to.

    Args:
        filter_root: Explicit root, or None to use this file system's own root.

    Returns:
        The root to relativize paths against.
    """
    raise NotImplementedError()

Node dataclass

Node(
    path_part,
    parent=None,
    children=dict(),
    size_bytes=0,
    object_count=0,
    last_modified=BEGINNING_OF_TIME,
    is_path_part_prefix=False,
    is_path_part_suffix=False,
)

Represents an object or folder in an file system path.

Attributes:

Name	Type	Description
path_part	str	Specifies the key part of the fs path (an edge) to this node.
parent	Node | None	Optionally specify the parent node to which this node is connected. By default, this is None.
children	dict[str, Node]	Child nodes that exist under this path prefix.
size_bytes	int	The size (in bytes) of all objects under this path prefix.
object_count	int	The number of objects under this path prefix.
last_modified	datetime	The most recent date any objects under this prefix were last modified.

S3FileSystem dataclass

S3FileSystem(bucket, key)

Bases: BaseFileSystem

Generates a FS tree structure of an S3 path with size and object count stats.

Attributes:

Name	Type	Description
bucket	str	The S3 bucket to describe.
key	str	The S3 key to describe.

get_file_system

get_file_system(path, filter_config=None, filter_root=None)

Build the file system tree appropriate to the given path.

Parameters:

Name	Type	Description	Default
path	str | Path	An S3 path, EFS path, or local path.	required
filter_config	DataSyncFilterConfig | None	Optional include/exclude filters restricting the tree to matching objects.	None
filter_root	str | None	Root that patterns are matched relative to. Defaults to path.	None

Returns:

Type	Description
BaseFileSystem	The refreshed file system.

Source code in src/aibs_informatics_aws_utils/data_sync/file_system.py

def get_file_system(
    path: str | Path,
    filter_config: DataSyncFilterConfig | None = None,
    filter_root: str | None = None,
) -> BaseFileSystem:
    """Build the file system tree appropriate to the given path.

    Args:
        path: An S3 path, EFS path, or local path.
        filter_config: Optional include/exclude filters restricting the tree to
            matching objects.
        filter_root: Root that patterns are matched relative to. Defaults to
            ``path``.

    Returns:
        The refreshed file system.
    """
    if isinstance(path, str) and S3Path.is_valid(path):
        return S3FileSystem.from_path(path, filter_config=filter_config, filter_root=filter_root)
    elif isinstance(path, str) and EFSPath.is_valid(path):
        return EFSFileSystem.from_path(path, filter_config=filter_config, filter_root=filter_root)
    else:
        return LocalFileSystem.from_path(
            path, filter_config=filter_config, filter_root=filter_root
        )