Filesystems

lume_services.services.files.filesystems.filesystem

Classes

Filesystem

Bases: ABC, BaseModel

Attributes

identifier class-attribute

identifier: str

Functions

dir_exists abstractmethod

dir_exists(dir: str, create_dir: bool = False) -> bool

Check that a directory exists

Parameters:

Name	Type	Description	Default
dir	str	Path of directory	required
create_dir	bool	Whether to create directory if it does not exist	False

Returns:

Type	Description
bool	bool

Source code in lume_services/services/files/filesystems/filesystem.py

@abstractmethod
def dir_exists(self, dir: str, create_dir: bool = False) -> bool:
    """Check that a directory exists

    Args:
        dir (str): Path of directory
        create_dir (bool): Whether to create directory if it does not exist

    Returns:
        bool
    """
    ...

file_exists abstractmethod

file_exists(filepath: str) -> bool

Check that a file exists

Parameters:

Name	Type	Description	Default
filepath	str	Path to file	required

Returns:

Type	Description
bool	bool

Source code in lume_services/services/files/filesystems/filesystem.py

@abstractmethod
def file_exists(self, filepath: str) -> bool:
    """Check that a file exists

    Args:
        filepath (str): Path to file

    Returns:
        bool

    """
    ...

create_dir abstractmethod

create_dir(dir: str) -> None

Create a directory on the filesystem.

Parameters:

Name	Type	Description	Default
dir	str	Directory path to create	required

Source code in lume_services/services/files/filesystems/filesystem.py

@abstractmethod
def create_dir(self, dir: str) -> None:
    """Create a directory on the filesystem.

    Args:
        dir (str): Directory path to create

    """
    ...

read abstractmethod

read(filepath: str, serializer: SerializerBase) -> Any

Read file from the filesystem.

Parameters:

Name	Type	Description	Default
filepath	str	Path of file to read	required
serializer	SerializerBase	Implementation of lume-base SerializerBase abstract base class	required

Source code in lume_services/services/files/filesystems/filesystem.py

@abstractmethod
def read(self, filepath: str, serializer: SerializerBase) -> Any:
    """Read file from the filesystem.

    Args:
        filepath (str): Path of file to read
        serializer (SerializerBase): Implementation of lume-base SerializerBase
            abstract base class

    """
    ...

write abstractmethod

write(
    filepath: str,
    object: Any,
    serializer: SerializerBase,
    create_dir: bool = False,
) -> None

Write a file to the filesystem.

Parameters:

Name	Type	Description	Default
filepath	str		required
object	Any	Object to serialize	required
serializer	SerializerBase	Implementation of lume-base SerializerBase abstract base class	required
create_dir	bool	Whether to create directory in case not implemented	False

Source code in lume_services/services/files/filesystems/filesystem.py

@abstractmethod
def write(
    self,
    filepath: str,
    object: Any,
    serializer: SerializerBase,
    create_dir: bool = False,
) -> None:
    """Write a file to the filesystem.

    Args:
        filepath (str):
        object (Any): Object to serialize
        serializer (SerializerBase): Implementation of lume-base SerializerBase
            abstract base class
        create_dir (bool): Whether to create directory in case not implemented

    """
    ...

lume_services.services.files.filesystems.local

Classes

LocalFilesystem

Bases: Filesystem

Handler for local filesystem.

Attributes

identifier class-attribute

identifier: str = 'local'

Functions

dir_exists

dir_exists(dir: str, create_dir: bool = False) -> bool

Check that a directory exists on the local filesystem.

Parameters:

Name	Type	Description	Default
dir	str	Path of directory	required
create_dir	bool	Whether to create directory if it does not exist	False

Returns:

Type	Description
bool	bool

Source code in lume_services/services/files/filesystems/local.py

def dir_exists(self, dir: str, create_dir: bool = False) -> bool:
    """Check that a directory exists on the local filesystem.

    Args:
        dir (str): Path of directory
        create_dir (bool): Whether to create directory if it does not exist

    Returns:
        bool

    """
    # use absolute path
    path = os.path.abspath(dir)
    if os.path.isdir(path):
        logger.info("Found directory %s on local filesystem.", path)
        return True

    # if creating...
    if create_dir:
        self.create_dir(path)
        return True
    else:
        logger.info("Unable to find directory %s on local filesystem.", path)
        return False

file_exists

file_exists(filepath: str) -> bool

Check that a file exists on the local filesystem.

Parameters:

Name	Type	Description	Default
filepath	str	Path to file	required

Returns:

Type	Description
bool	bool

Source code in lume_services/services/files/filesystems/local.py

def file_exists(self, filepath: str) -> bool:
    """Check that a file exists on the local filesystem.

    Args:
        filepath (str): Path to file

    Returns:
        bool

    """
    path = os.path.abspath(filepath)
    if os.path.isfile(path):
        logger.info("Found file %s on local filesystem.", path)
        return True

    else:
        logger.info("Unable to find file %s on local filesystem.", path)
        return False

create_dir

create_dir(dir: str) -> None

Create a directory on the local filesystem.

Parameters:

Name	Type	Description	Default
dir	str	Directory path to create	required

Source code in lume_services/services/files/filesystems/local.py

def create_dir(self, dir: str) -> None:
    """Create a directory on the local filesystem.

    Args:
        dir (str): Directory path to create

    """
    try:
        os.makedirs(dir, exist_ok=False)
    except Exception as e:
        logger.error("Unable to create directory %s on local filesystem.", dir)
        raise e

read

read(filepath: str, serializer: SerializerBase) -> Any

Read file from the local filesystem.

Parameters:

Name	Type	Description	Default
filepath	str	Path of file to read.	required
serializer	SerializerBase	Implementation of lume-base SerializerBase abstract base class.	required

Source code in lume_services/services/files/filesystems/local.py

def read(self, filepath: str, serializer: SerializerBase) -> Any:
    """Read file from the local filesystem.

    Args:
        filepath (str): Path of file to read.
        serializer (SerializerBase): Implementation of lume-base SerializerBase
            abstract base class.

    """
    path = os.path.abspath(filepath)
    content = serializer.deserialize(path)
    return content

write

write(
    filepath: str,
    object: Any,
    serializer: SerializerBase,
    create_dir: bool = False,
) -> None

Write a file to the local filesystem.

Parameters:

Name	Type	Description	Default
filepath	str		required
object	Any	Object to serialize	required
serializer	SerializerBase	Implementation of lume-base SerializerBase abstract base class	required
create_dir	bool	Whether to create directory in case not implemented	False

Source code in lume_services/services/files/filesystems/local.py

def write(
    self,
    filepath: str,
    object: Any,
    serializer: SerializerBase,
    create_dir: bool = False,
) -> None:
    """Write a file to the local filesystem.

    Args:
        filepath (str):
        object (Any): Object to serialize
        serializer (SerializerBase): Implementation of lume-base SerializerBase
            abstract base class
        create_dir (bool): Whether to create directory in case not implemented

    """
    path = os.path.abspath(filepath)
    dir = os.path.dirname(path)

    if create_dir and not self.dir_exists(dir):
        self.create_dir(dir)

    serializer.serialize(path, object)

lume_services.services.files.filesystems.mounted

Classes

PathNotInMount

PathNotInMount(
    filesystem_identifier: str,
    path: str,
    mount_path: str,
    mount_alias: str,
)

Bases: Exception

Source code in lume_services/services/files/filesystems/mounted.py

def __init__(
    self, filesystem_identifier: str, path: str, mount_path: str, mount_alias: str
):
    self.filesystem_identifier = filesystem_identifier
    self.path = (path,)
    self.mount_path = mount_path
    self.mount_alias = mount_alias
    self.message = "Path %s not in mount for mounted filesystem identifier: %s, \
        Mount path: %s, Mount alias: %s"
    super().__init__(
        self.message,
        self.path,
        self.filesystem_identifier,
        self.mount_path,
        self.mount_alias,
    )

Attributes

filesystem_identifier instance-attribute

filesystem_identifier = filesystem_identifier

path instance-attribute

path = (path)

mount_path instance-attribute

mount_path = mount_path

mount_alias instance-attribute

mount_alias = mount_alias

message instance-attribute

message = "Path %s not in mount for mounted filesystem identifier: %s,             Mount path: %s, Mount alias: %s"

Functions

MountedFilesystem

Bases: LocalFilesystem

Handler for mounted filesystem. Modifies the LocalFilesystem to implements checks for mount path modifications. Container and container orchestration tools often allow the ability to provide an alias for a mounted directory. This handler accounts for the mount base and verifies that the file is within the path.

Attributes

identifier class-attribute

identifier: str = 'mounted'

mount_path class-attribute

mount_path: str

mount_alias class-attribute

mount_alias: str

mount_type class-attribute

mount_type: _HostMountLiteral = 'DirectoryOrCreate'

Functions

validate_mount_path

validate_mount_path(v, values)

Source code in lume_services/services/files/filesystems/mounted.py

@validator("mount_path", pre=True)
def validate_mount_path(cls, v, values):
    mount_type = values.get("mount_type")

    if mount_type == "DirectoryOrCreate":
        os.mkdir(v)

    return v

dir_exists

dir_exists(dir: str, create_dir: bool = False) -> bool

Check that a directory exists on the mounted filesystem.

Parameters:

Name	Type	Description	Default
dir	str	Path of directory	required
create_dir	bool	Whether to create directory if it does not exist	False

Returns:

Type	Description
bool	bool

Source code in lume_services/services/files/filesystems/mounted.py

def dir_exists(self, dir: str, create_dir: bool = False) -> bool:
    """Check that a directory exists on the mounted filesystem.

    Args:
        dir (str): Path of directory
        create_dir (bool): Whether to create directory if it does not exist

    Returns:
        bool

    """
    dir = self._check_mounted_path(dir)
    return super().dir_exists(dir, create_dir=create_dir)

file_exists

file_exists(filepath: str) -> bool

Check that a file exists on the mounted filesystem.

Parameters:

Name	Type	Description	Default
filepath	str	Path to file	required

Returns:

Type	Description
bool	bool

Source code in lume_services/services/files/filesystems/mounted.py

def file_exists(self, filepath: str) -> bool:
    """Check that a file exists on the mounted filesystem.

    Args:
        filepath (str): Path to file

    Returns:
        bool

    """

    filepath = self._check_mounted_path(filepath)
    return super().file_exists(filepath)

create_dir

create_dir(dir: str) -> None

Create a directory on the mounted filesystem.

Parameters:

Name	Type	Description	Default
dir	str	Directory path to create	required

Source code in lume_services/services/files/filesystems/mounted.py

def create_dir(self, dir: str) -> None:
    """Create a directory on the mounted filesystem.

    Args:
        dir (str): Directory path to create

    """
    dir = self._check_mounted_path(dir)
    super().create_dir(dir)

read

read(filepath: str, serializer: SerializerBase) -> Any

Read file from the mounted filesystem.

Parameters:

Name	Type	Description	Default
filepath	str	Path of file to read	required
serializer	SerializerBase	Implementation of lume-base SerializerBase abstract base class	required

Source code in lume_services/services/files/filesystems/mounted.py

def read(self, filepath: str, serializer: SerializerBase) -> Any:
    """Read file from the mounted filesystem.

    Args:
        filepath (str): Path of file to read
        serializer (SerializerBase): Implementation of lume-base SerializerBase
            abstract base class

    """
    filepath = self._check_mounted_path(filepath)
    return super().read(filepath, serializer)

write

write(
    filepath: str,
    object: Any,
    serializer: SerializerBase,
    create_dir: bool = False,
) -> None

Write a file to the mounted filesystem.

Parameters:

Name	Type	Description	Default
filepath	str		required
object	Any	Object to serialize	required
serializer	SerializerBase	Implementation of lume-base SerializerBase abstract base class	required
create_dir	bool	Whether to create directory in case not implemented	False

Source code in lume_services/services/files/filesystems/mounted.py

def write(
    self,
    filepath: str,
    object: Any,
    serializer: SerializerBase,
    create_dir: bool = False,
) -> None:
    """Write a file to the mounted filesystem.

    Args:
        filepath (str):
        object (Any): Object to serialize
        serializer (SerializerBase): Implementation of lume-base SerializerBase
            abstract base class
        create_dir (bool): Whether to create directory in case not implemented

    """
    filepath = self._check_mounted_path(filepath)

    super().write(filepath, object, serializer, create_dir=create_dir)