Skip to content

fs

mlte/store/catalog/underlying/fs.py

Implementation of local file system catalog store.

FileSystemCatalogEntryMapper

Bases: CatalogEntryMapper

FS mapper for the catalog entry resource.

Source code in mlte/store/catalog/underlying/fs.py 
 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
class FileSystemCatalogEntryMapper(CatalogEntryMapper):
    """FS mapper for the catalog entry resource."""

    ENTRIES_FOLDER = "entries"
    """Subfolder for entries."""

    def __init__(self, storage: FileSystemStorage) -> None:
        self.storage = storage.clone()
        """A reference to underlying storage."""

        self.storage.setup_base_path(
            Path(self.storage.sub_folder, self.ENTRIES_FOLDER)
        )
        """Set the subfolder for this resource."""

    def create(self, entry: CatalogEntry, context: Any = None) -> CatalogEntry:
        self.storage.ensure_resource_does_not_exist(entry.header.identifier)
        return self._write_entry(entry)

    def edit(self, entry: CatalogEntry, context: Any = None) -> CatalogEntry:
        self.storage.ensure_resource_exists(entry.header.identifier)
        return self._write_entry(entry)

    def read(self, entry_id: str, context: Any = None) -> CatalogEntry:
        return self._read_entry(entry_id)

    def list(self, context: Any = None) -> List[str]:
        return self.storage.list_resources()

    def delete(self, entry_id: str, context: Any = None) -> CatalogEntry:
        self.storage.ensure_resource_exists(entry_id)
        entry = self._read_entry(entry_id)
        self.storage.delete_resource(entry_id)
        return entry

    def _read_entry(self, entry_id: str) -> CatalogEntry:
        """Reads a catalog entry."""
        self.storage.ensure_resource_exists(entry_id)
        return CatalogEntry(**self.storage.read_resource(entry_id))

    def _write_entry(self, entry: CatalogEntry) -> CatalogEntry:
        """Writes an entry to storage."""
        self.storage.write_resource(entry.header.identifier, entry.to_json())
        return self._read_entry(entry.header.identifier)

ENTRIES_FOLDER = 'entries' class-attribute instance-attribute

Subfolder for entries.

storage = storage.clone() instance-attribute

A reference to underlying storage.

FileSystemCatalogStore

Bases: CatalogStore

A local file system implementation of the MLTE catalog store.

Source code in mlte/store/catalog/underlying/fs.py 
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
class FileSystemCatalogStore(CatalogStore):
    """A local file system implementation of the MLTE catalog store."""

    BASE_CATALOGS_FOLDER = "catalogs"
    """Base folder to store catalog entries in."""

    DEFAULT_CATALOG_FOLDER = "catalog"
    """A default name for a catalog folder."""

    def __init__(
        self, uri: StoreURI, catalog_folder: Optional[str] = None
    ) -> None:
        self.uri = uri
        """Store uri."""

        if not catalog_folder:
            catalog_folder = self.DEFAULT_CATALOG_FOLDER

        catalog_folder = os.path.join(self.BASE_CATALOGS_FOLDER, catalog_folder)
        self.storage = FileSystemStorage(uri=uri, sub_folder=catalog_folder)
        """The underlying storage for the store."""

    def session(self) -> FileSystemCatalogStoreSession:
        """
        Return a session handle for the store instance.
        :return: The session handle
        """
        return FileSystemCatalogStoreSession(
            storage=self.storage, read_only=self.read_only
        )

BASE_CATALOGS_FOLDER = 'catalogs' class-attribute instance-attribute

Base folder to store catalog entries in.

DEFAULT_CATALOG_FOLDER = 'catalog' class-attribute instance-attribute

A default name for a catalog folder.

storage = FileSystemStorage(uri=uri, sub_folder=catalog_folder) instance-attribute

The underlying storage for the store.

uri = uri instance-attribute

Store uri.

session()

Return a session handle for the store instance.

Returns:

Type Description
FileSystemCatalogStoreSession

The session handle
Source code in mlte/store/catalog/underlying/fs.py 
49
50
51
52
53
54
55
56
def session(self) -> FileSystemCatalogStoreSession:
    """
    Return a session handle for the store instance.
    :return: The session handle
    """
    return FileSystemCatalogStoreSession(
        storage=self.storage, read_only=self.read_only
    )

FileSystemCatalogStoreSession

Bases: CatalogStoreSession

A local file-system implementation of the MLTE catalog store.

Source code in mlte/store/catalog/underlying/fs.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
class FileSystemCatalogStoreSession(CatalogStoreSession):
    """A local file-system implementation of the MLTE catalog store."""

    def __init__(
        self, storage: FileSystemStorage, read_only: bool = False
    ) -> None:
        self.storage = storage
        """The storage."""

        self.read_only = read_only
        """Whether this is read only or not."""

        self.entry_mapper = FileSystemCatalogEntryMapper(storage=storage)
        """The mapper to entries CRUD."""

    def close(self) -> None:
        """Close the session."""
        # Closing a local FS session is a no-op.
        pass

entry_mapper = FileSystemCatalogEntryMapper(storage=storage) instance-attribute

The mapper to entries CRUD.

read_only = read_only instance-attribute

Whether this is read only or not.

storage = storage instance-attribute

The storage.

close()

Close the session.

Source code in mlte/store/catalog/underlying/fs.py 
79
80
81
82
def close(self) -> None:
    """Close the session."""
    # Closing a local FS session is a no-op.
    pass