Mount Point¶

EFS mount point management utilities.

MountPointConfiguration `dataclass` ¶

MountPointConfiguration(
    file_system, access_point, mount_point
)

Adapter for translating and mapping paths between AWS Elastic File System (EFS) mount points and local file system paths.

local file systems are most often used by: - EC2 instances - AWS Lambda functions

This class provides functionality to adapt file paths for applications that interact with AWS EFS. It allows for the conversion of paths from the EFS perspective to the local file system perspective and vice versa, considering the mount point and access point configurations.

Attributes:

Name	Type	Description
`file_system`	`FileSystemDescriptionTypeDef`	The description of the EFS file system.
`access_point`	`Optional[AccessPointDescriptionTypeDef]`	The description of the EFS access point, if used.
`mount_point`	`Path`	The local file system path where the EFS is mounted.

access_point_path `property` ¶

access_point_path

This is the access point on file system mounted to host

is_writeable `property` ¶

is_writeable

Returns true, if the mount point is writeable

mount_point_path `property` ¶

mount_point_path

This is the path on the host where the access point will be mounted

as_efs_path ¶

as_efs_path(path)

Converts a path to a path on the EFS file system.

Behavior: - If the path is absolute, the path is made relative to either the mount point or access point first. examples: - "/efs/accesspoint/path/to/file" -> "path/to/file" - "/mnt/efs/path/to/file" -> "path/to/file" - If the path is absolute and does not start with the mount point or access point, a ValueError is raised. - If the path is relative, It is assumed to be the relative path from the access point. e.g. "path/to/file" -> "/efs/accesspoint/path/to/file"

Parameters:

Name	Type	Description	Default
`path`	`StrPath`	The path to convert to a path on the EFS file system.	required

Returns:

Type	Description
`Path`	absolute path on the EFS file system.

Source code in src/aibs_informatics_aws_utils/efs/mount_point.py

def as_efs_path(self, path: StrPath) -> Path:
    """
    Converts a path to a path on the EFS file system.

    Behavior:
    - If the path is absolute, the path is made relative to either the
        mount point or access point first. examples:
        - "/efs/accesspoint/path/to/file" -> "path/to/file"
        - "/mnt/efs/path/to/file" -> "path/to/file"
    - If the path is absolute and does not start with the mount point or access point,
        a ValueError is raised.
    - If the path is relative, It is assumed to be the relative path from the access point.
        e.g. "path/to/file" -> "/efs/accesspoint/path/to/file"

    Args:
        path (StrPath): The path to convert to a path on the EFS file system.

    Returns:
        absolute path on the EFS file system.
    """
    path = self.as_relative_path(path)
    return (self.access_point_path / path).resolve()

as_efs_uri ¶

as_efs_uri(path)

Converts a path to a customized URI EFSPath describing the location on an EFS file system.

Parameters:

Name	Type	Description	Default
`path`	`StrPath`	The path to convert to a path on the EFS file system.	required

Returns:

Type	Description
`EFSPath`	absolute URI path on the EFS file system.

Source code in src/aibs_informatics_aws_utils/efs/mount_point.py

def as_efs_uri(self, path: StrPath) -> EFSPath:
    """
    Converts a path to a customized URI EFSPath describing the location on an EFS file system.

    Args:
        path (StrPath): The path to convert to a path on the EFS file system.

    Returns:
        absolute URI path on the EFS file system.
    """
    return EFSPath.build(
        resource_id=self.file_system["FileSystemId"],
        path=self.as_efs_path(path),
    )

as_env_vars ¶

as_env_vars(name=None)

Converts the mount point configuration to environment variables.

Source code in src/aibs_informatics_aws_utils/efs/mount_point.py

def as_env_vars(self, name: Optional[str] = None) -> Dict[str, str]:
    """Converts the mount point configuration to environment variables."""
    if self.access_point and self.access_point.get("AccessPointId"):
        mount_point_id = self.access_point["AccessPointId"]
    else:
        mount_point_id = self.file_system["FileSystemId"]

    return self.to_env_vars(self.mount_point, mount_point_id, name)

as_mounted_path ¶

as_mounted_path(path)

Converts a path to a path on the host where the EFS is mounted.

Behavior: - If the path is absolute, the path is made relative to either the mount point or access point first. examples: - "/efs/accesspoint/path/to/file" -> "path/to/file" - "/mnt/efs/path/to/file" -> "path/to/file" - If the path is absolute and does not start with the mount point or access point, a ValueError is raised. - If the path is relative, It is assumed to be the relative path from the mount point. e.g. "path/to/file" -> "/mnt/efs/path/to/file"

Parameters:

Name	Type	Description	Default
`path`	`StrPath`	The path to convert to a path on the host where the EFS is mounted.	required

Returns:

Type	Description
`Path`	absolute path on the host where the EFS is mounted.

Source code in src/aibs_informatics_aws_utils/efs/mount_point.py

def as_mounted_path(self, path: StrPath) -> Path:
    """
    Converts a path to a path on the host where the EFS is mounted.

    Behavior:
    - If the path is absolute, the path is made relative to either the mount point
        or access point first. examples:
        - "/efs/accesspoint/path/to/file" -> "path/to/file"
        - "/mnt/efs/path/to/file" -> "path/to/file"
    - If the path is absolute and does not start with the mount point or access point,
        a ValueError is raised.
    - If the path is relative, It is assumed to be the relative path from the mount point.
        e.g. "path/to/file" -> "/mnt/efs/path/to/file"

    Args:
        path (StrPath): The path to convert to a path on the host where the EFS is mounted.

    Returns:
        absolute path on the host where the EFS is mounted.
    """
    path = self.as_relative_path(path)
    return (self.mount_point / path).resolve()

as_relative_path ¶

as_relative_path(path)

Converts a path to a relative path from the mount point or access point.

Behavior: - If the path is absolute and is relative to the mount point or access point, it is returned as a relative path. - If the path is already relative, it is returned as is. - If the path is absolute and is not relative to the mount point or access point, a ValueError is raised.

Parameters:

Name	Type	Description	Default
`path`	`StrPath`	The path to convert to a relative path.	required

Returns:

Type	Description
`Path`	The relative path.

Source code in src/aibs_informatics_aws_utils/efs/mount_point.py

def as_relative_path(self, path: StrPath) -> Path:
    """
    Converts a path to a relative path from the mount point or access point.

    Behavior:
    - If the path is absolute and is relative to the mount point or access point,
        it is returned as a relative path.
    - If the path is already relative,
        it is returned as is.
    - If the path is absolute and is not relative to the mount point or access point,
        a ValueError is raised.


    Args:
        path (StrPath): The path to convert to a relative path.

    Returns:
        The relative path.

    """
    path = Path(path)
    if path.is_absolute():
        # Check longer path first. This is to avoid the case where the mount point
        # is a subdirectory of the access point.
        check_mount_path_first = len(self.mount_point_path.as_posix()) > len(
            self.access_point_path.as_posix()
        )
        if check_mount_path_first and self.is_mounted_path(path):
            path = path.relative_to(self.mount_point)
        elif self.is_efs_path(path):
            path = path.relative_to(self.access_point_path)
        elif self.is_mounted_path(path):
            path = path.relative_to(self.mount_point)
        else:
            raise ValueError(
                f"Path {path} is not relative to either the mount point {self.mount_point} "
                f"or the access point {self.access_point_path}"
            )
    return path

build `classmethod` ¶

build(
    mount_point,
    access_point=None,
    file_system=None,
    access_point_tags=None,
    file_system_tags=None,
)

Creates a new config from the given mount point and access point or file system.

Important: Must provide either access point or file system.

Parameters:

Name	Type	Description	Default
`mount_point`	`StrPath`	Intended mount point of the EFS file system on the host.	required
`access_point`	`Optional[Union[str, AccessPointDescriptionTypeDef]]`	Identifier of the access point or the access point description. If specified as a string, can be either the access point id or name. Defaults to None.	`None`
`file_system`	`Optional[Union[str, FileSystemDescriptionTypeDef]]`	Identifier of the file system or the file system description. If specified as a string, can be either the file system id or name. Defaults to None.	`None`
`access_point_tags`	`Optional[Dict[str, str]]`	Tags to filter the access point. Defaults to None.	`None`
`file_system_tags`	`Optional[Dict[str, str]]`	Tags to filter the file system. Defaults to None.	`None`

Raises: ValueError: if neither access point nor file system is provided.

Returns:

Type	Description
`MountPointConfiguration`	The config

Source code in src/aibs_informatics_aws_utils/efs/mount_point.py

@classmethod
def build(
    cls,
    mount_point: StrPath,
    access_point: Optional[Union[str, AccessPointDescriptionTypeDef]] = None,
    file_system: Optional[Union[str, FileSystemDescriptionTypeDef]] = None,
    access_point_tags: Optional[Dict[str, str]] = None,
    file_system_tags: Optional[Dict[str, str]] = None,
) -> MountPointConfiguration:
    """Creates a new config from the given mount point and access point or file system.

    Important: Must provide either access point or file system.

    Args:
        mount_point (StrPath): Intended mount point of the EFS file system on the host.
        access_point (Optional[Union[str, AccessPointDescriptionTypeDef]]):
            Identifier of the access point or the access point description.
            If specified as a string, can be either the access point id or name.
            Defaults to None.
        file_system (Optional[Union[str, FileSystemDescriptionTypeDef]]):
            Identifier of the file system or the file system description.
            If specified as a string, can be either the file system id or name.
            Defaults to None.
        access_point_tags (Optional[Dict[str, str]]): Tags to filter the access point.
            Defaults to None.
        file_system_tags (Optional[Dict[str, str]]): Tags to filter the file system.
            Defaults to None.
    Raises:
        ValueError: if neither access point nor file system is provided.

    Returns:
        The config
    """

    if access_point is None and file_system is None:
        raise ValueError("Either access point or file system must be provided")

    file_system_id = None
    if isinstance(file_system, str):
        logger.info(f"Resolving file system config from name {file_system}")
        fs_id, fs_name = (
            (FileSystemId(file_system).normalized, None)
            if FileSystemId.is_valid(file_system)
            else (None, file_system)
        )
        file_system = get_efs_file_system(
            name=fs_name, file_system_id=fs_id, tags=file_system_tags
        )
        file_system_id = file_system["FileSystemId"]
        logger.info(f"Resolved file system id {file_system_id}")

    access_point_id = None
    if isinstance(access_point, str):
        logger.info(f"Resolving access point config from name {access_point}")
        ap_id, ap_name = (
            (AccessPointId(access_point).normalized, None)
            if AccessPointId.is_valid(access_point)
            else (None, access_point)
        )
        access_point = get_efs_access_point(
            access_point_name=ap_name,
            access_point_id=ap_id,
            file_system_id=file_system_id,
            access_point_tags=access_point_tags,
        )
        access_point_id = access_point.get("AccessPointId")
        logger.info(f"Resolved access point id {access_point_id}")

    if file_system is None:
        assert access_point is not None
        file_system_id = access_point.get("FileSystemId")
        logger.info(f"Resolving file system config from access point id {access_point_id}")
        file_system = get_efs_file_system(file_system_id=access_point.get("FileSystemId"))
        logger.info(f"Resolved file system id {file_system_id}")

    if not isinstance(mount_point, Path):
        mount_point = Path(mount_point)

    logger.info(
        f"Generating EFS Mount Point Connection using "
        f"file system {file_system}, "
        f"access point {access_point}, "
        f"with target {mount_point} mount point"
    )
    return cls(
        file_system=file_system,
        access_point=access_point,
        mount_point=mount_point,
    )

is_efs_path ¶

is_efs_path(path)

Checks if the path is relative to the access point

Source code in src/aibs_informatics_aws_utils/efs/mount_point.py

def is_efs_path(self, path: StrPath) -> bool:
    """Checks if the path is relative to the access point"""
    return Path(path).is_relative_to(self.access_point_path)

is_mounted_path ¶

is_mounted_path(path)

Checks if the path is relative to the mount point

Source code in src/aibs_informatics_aws_utils/efs/mount_point.py

def is_mounted_path(self, path: StrPath) -> bool:
    """Checks if the path is relative to the mount point"""
    return Path(path).is_relative_to(self.mount_point)

to_env_vars `classmethod` ¶

to_env_vars(mount_point, mount_point_id, name=None)

Converts the mount point configuration to environment variables.

Parameters:

Name	Type	Description	Default
`mount_point`	`StrPath`	The mount point.	required
`mount_point_id`	`Union[str, AccessPointId, FileSystemId]`	The mount point identifier.	required
`name`	`Optional[str]`	Optional name for the mount point. Defaults to None.	`None`

Returns:

Type	Description
`Dict[str, str]`	Dict[str, str]: The environment variables.

Source code in src/aibs_informatics_aws_utils/efs/mount_point.py

@classmethod
def to_env_vars(
    cls,
    mount_point: StrPath,
    mount_point_id: Union[str, AccessPointId, FileSystemId],
    name: Optional[str] = None,
) -> Dict[str, str]:
    """Converts the mount point configuration to environment variables.

    Args:
        mount_point (StrPath): The mount point.
        mount_point_id (Union[str, AccessPointId, FileSystemId]): The mount point identifier.
        name (Optional[str]): Optional name for the mount point. Defaults to None.

    Returns:
        Dict[str, str]: The environment variables.
    """

    if not isinstance(mount_point, Path):
        mount_point = Path(mount_point)

    if not isinstance(mount_point_id, AccessPointId) and not isinstance(
        mount_point_id, FileSystemId
    ):
        if AccessPointId.is_valid(mount_point_id):
            mount_point_id = AccessPointId(mount_point_id).normalized
        elif FileSystemId.is_valid(mount_point_id):
            mount_point_id = FileSystemId(mount_point_id).normalized
        else:
            raise ValueError(
                f"Invalid mount point id {mount_point_id}. "
                "Must be either an access point id or file system id."
            )
    else:
        mount_point_id = mount_point_id.normalized

    if name and not re.match(r"[a-zA-Z][\w]{0,11}", name):
        raise ValueError(
            f"Invalid mount point name {name}. Must be a valid environment variable name."
        )
    else:
        name = name or sha256_hexdigest([mount_point.as_posix(), mount_point_id])[:6]

    return {
        f"{EFS_MOUNT_POINT_PATH_VAR}_{name}": mount_point.as_posix(),
        f"{EFS_MOUNT_POINT_ID_VAR}_{name}": mount_point_id,
    }

translate_mounted_path ¶

translate_mounted_path(path, other)

Translates the location of a path described by another mount point to a location on this mount point

Parameters:

Name	Type	Description	Default
`path`	`StrPath`	The path to translate. should be relative path or absolute relative to the other mount point/access point.	required
`other`	`T`	The config of other mount point/access point to translate from.	required

Raises:

Type	Description
`ValueError`	If file system of other mount point/access point is not the same as this mount point/access point.

Returns:

Type	Description
`Path`	The translated absolute path on this mount point.

Source code in src/aibs_informatics_aws_utils/efs/mount_point.py

def translate_mounted_path(self, path: StrPath, other: MountPointConfiguration) -> Path:
    """
    Translates the location of a path described by another mount point to
    a location on this mount point

    Args:
        path (StrPath): The path to translate. should be relative path
            or absolute relative to the other mount point/access point.
        other (T): The config of other mount point/access point to translate from.

    Raises:
        ValueError: If file system of other mount point/access point is
            not the same as this mount point/access point.

    Returns:
        The translated absolute path on this mount point.
    """
    path = Path(path)
    if self.file_system["FileSystemId"] != other.file_system["FileSystemId"]:
        raise ValueError(
            f"Cannot resolve path {path} from file system "
            f"{other.file_system['FileSystemId']} to file system "
            f"{self.file_system['FileSystemId']}. They are not the same file system!"
        )

    # First, convert the path to a global path on EFS File System
    global_path = other.as_efs_path(path)
    # Next, convert the global path to the path on the container
    return self.as_mounted_path(global_path)

deduplicate_mount_points ¶

deduplicate_mount_points(mount_points)

Deduplicates a list of MountPointConfiguration objects based on the file system id and access point id.

Source code in src/aibs_informatics_aws_utils/efs/mount_point.py

def deduplicate_mount_points(
    mount_points: List[MountPointConfiguration],
) -> List[MountPointConfiguration]:
    """
    Deduplicates a list of MountPointConfiguration objects based on the file
    system id and access point id.
    """

    unique_configs: Dict[str, MountPointConfiguration] = {}
    for mp_config in mount_points:
        key = mp_config.mount_point.as_posix()

        if key in unique_configs:
            other = unique_configs[key]
            if mp_config.access_point_path != other.access_point_path:
                raise ValueError(
                    f"Found conflicting mount points for {key}: "
                    f"{mp_config.access_point_path} and {unique_configs[key].access_point_path}"
                )
            elif mp_config.file_system["FileSystemId"] != other.file_system["FileSystemId"]:
                raise ValueError(
                    f"Found conflicting file systems for {key}: "
                    f"{mp_config.file_system['FileSystemId']} "
                    f"and {unique_configs[key].file_system['FileSystemId']}"
                )
            elif mp_config.access_point != other.access_point:
                raise ValueError(
                    f"Found conflicting access points for {key}: "
                    f"{mp_config.access_point} and {unique_configs[key].access_point}"
                )
            continue
        unique_configs[key] = mp_config
    return list(unique_configs.values())

Mount Point¶

MountPointConfiguration dataclass ¶

access_point_path property ¶

is_writeable property ¶

mount_point_path property ¶

as_efs_path ¶

as_efs_uri ¶

as_env_vars ¶

as_mounted_path ¶

as_relative_path ¶

build classmethod ¶

is_efs_path ¶

is_mounted_path ¶

to_env_vars classmethod ¶

translate_mounted_path ¶

deduplicate_mount_points ¶

MountPointConfiguration `dataclass` ¶

access_point_path `property` ¶

is_writeable `property` ¶

mount_point_path `property` ¶

build `classmethod` ¶

to_env_vars `classmethod` ¶