Skip to content

Filesystems

Filesystem

Bases: ABC, BaseModel

Source code in lume_services/services/files/filesystems/filesystem.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
class Filesystem(ABC, BaseModel):
    identifier: str

    @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
        """
        ...

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

        Args:
            filepath (str): Path to file

        Returns:
            bool

        """
        ...

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

        Args:
            dir (str): Directory path to create

        """
        ...

    @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

        """
        ...

    @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

        """
        ...

create_dir(dir) abstractmethod

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 
36
37
38
39
40
41
42
43
44
@abstractmethod
def create_dir(self, dir: str) -> None:
    """Create a directory on the filesystem.

    Args:
        dir (str): Directory path to create

    """
    ...

dir_exists(dir, create_dir=False) abstractmethod

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 
10
11
12
13
14
15
16
17
18
19
20
21
@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(filepath) abstractmethod

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 
23
24
25
26
27
28
29
30
31
32
33
34
@abstractmethod
def file_exists(self, filepath: str) -> bool:
    """Check that a file exists

    Args:
        filepath (str): Path to file

    Returns:
        bool

    """
    ...

read(filepath, serializer) abstractmethod

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 
46
47
48
49
50
51
52
53
54
55
56
@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(filepath, object, serializer, create_dir=False) abstractmethod

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 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
@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

    """
    ...

LocalFilesystem

Bases: Filesystem

Handler for local filesystem.

Source code in lume_services/services/files/filesystems/local.py 
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
class LocalFilesystem(Filesystem):
    """Handler for local filesystem."""

    identifier: str = "local"

    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

    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

    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

    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

    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)

create_dir(dir)

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 
60
61
62
63
64
65
66
67
68
69
70
71
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

dir_exists(dir, create_dir=False)

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 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
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(filepath)

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 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
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

read(filepath, serializer)

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 
73
74
75
76
77
78
79
80
81
82
83
84
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(filepath, object, serializer, create_dir=False)

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 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
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)

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.

Source code in lume_services/services/files/filesystems/mounted.py 
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
class MountedFilesystem(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.

    """

    identifier: str = "mounted"
    mount_path: str
    mount_alias: str
    mount_type: _HostMountLiteral = "DirectoryOrCreate"

    @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

    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)

    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)

    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)

    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)

    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)

    def _check_mounted_path(self, path: str):
        """Checks that the path exists inside the mount point relative to mount path or
            alias.

        Args:
            path (str): Path to validate

        """
        print(path)

        if self.mount_alias in path:
            return path

        elif self.mount_path in path:
            return path.replace(self.mount_path, self.mount_alias)

        else:
            raise PathNotInMount(
                self.identifier, path, self.mount_path, self.mount_alias
            )

create_dir(dir)

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 
74
75
76
77
78
79
80
81
82
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)

dir_exists(dir, create_dir=False)

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 
46
47
48
49
50
51
52
53
54
55
56
57
58
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(filepath)

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 
60
61
62
63
64
65
66
67
68
69
70
71
72
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)

read(filepath, serializer)

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 
84
85
86
87
88
89
90
91
92
93
94
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(filepath, object, serializer, create_dir=False)

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 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
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)