Skip to content

memory

mlte/store/catalog/underlying/memory.py

Implementation of in-memory catalog store.

InMemoryCatalogEntryMapper

Bases: CatalogEntryMapper

In-memory mapper for the catalog resource.

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

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

    def create(self, entry: CatalogEntry, context: Any = None) -> CatalogEntry:
        if entry.header.identifier in self.storage.entries:
            raise errors.ErrorAlreadyExists(f"Entry {entry.header.identifier}")

        self.storage.entries[entry.header.identifier] = entry
        return entry

    def edit(self, entry: CatalogEntry, context: Any = None) -> CatalogEntry:
        if entry.header.identifier not in self.storage.entries:
            raise errors.ErrorNotFound(f"Entry {entry.header.identifier}")

        self.storage.entries[entry.header.identifier] = entry
        return entry

    def read(self, entry_id: str, context: Any = None) -> CatalogEntry:
        if entry_id not in self.storage.entries:
            raise errors.ErrorNotFound(f"Entry {entry_id}")

        entry = self.storage.entries[entry_id]
        return entry

    def list(self, context: Any = None) -> List[str]:
        return [username for username in self.storage.entries.keys()]

    def delete(self, entry_id: str, context: Any = None) -> CatalogEntry:
        if entry_id not in self.storage.entries:
            raise errors.ErrorNotFound(f"Entry {entry_id}")

        popped = self.storage.entries[entry_id]
        del self.storage.entries[entry_id]
        return popped

storage = storage instance-attribute

A reference to underlying storage.

InMemoryCatalogStore

Bases: CatalogStore

An in-memory implementation of the MLTE user store.

Source code in mlte/store/catalog/underlying/memory.py 
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
class InMemoryCatalogStore(CatalogStore):
    """An in-memory implementation of the MLTE user store."""

    def __init__(self, uri: StoreURI) -> None:
        self.storage = MemoryCatalogStorage(uri)
        """The underlying storage for the store."""

        super().__init__(uri=uri)
        """Store uri."""

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

    def clone(self) -> InMemoryCatalogStore:
        """
        Clones the store. Shallow clone.
        :return: The cloned store
        """
        clone = InMemoryCatalogStore(self.uri)
        clone.storage.entries = self.storage.entries.copy()
        return clone

storage = MemoryCatalogStorage(uri) instance-attribute

The underlying storage for the store.

clone()

Clones the store. Shallow clone.

Returns:

Type Description
InMemoryCatalogStore

The cloned store
Source code in mlte/store/catalog/underlying/memory.py 
45
46
47
48
49
50
51
52
def clone(self) -> InMemoryCatalogStore:
    """
    Clones the store. Shallow clone.
    :return: The cloned store
    """
    clone = InMemoryCatalogStore(self.uri)
    clone.storage.entries = self.storage.entries.copy()
    return clone

session()

Return a session handle for the store instance.

Returns:

Type Description
CatalogStoreSession

The session handle
Source code in mlte/store/catalog/underlying/memory.py 
36
37
38
39
40
41
42
43
def session(self) -> CatalogStoreSession:
    """
    Return a session handle for the store instance.
    :return: The session handle
    """
    return InMemoryCatalogStoreSession(
        storage=self.storage, read_only=self.read_only
    )

InMemoryCatalogStoreSession

Bases: CatalogStoreSession

An in-memory implementation of the MLTE user store.

Source code in mlte/store/catalog/underlying/memory.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
class InMemoryCatalogStoreSession(CatalogStoreSession):
    """An in-memory implementation of the MLTE user store."""

    def __init__(
        self, *, storage: MemoryCatalogStorage, 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 = InMemoryCatalogEntryMapper(storage=storage)

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

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/memory.py 
83
84
85
86
def close(self) -> None:
    """Close the session."""
    # Closing an in-memory session is a no-op.
    pass

MemoryCatalogStorage

Bases: Storage

A simple storage wrapper for the in-memory store.

Source code in mlte/store/catalog/underlying/memory.py 
55
56
57
58
59
60
61
class MemoryCatalogStorage(Storage):
    """A simple storage wrapper for the in-memory store."""

    def __init__(self, uri: StoreURI) -> None:
        super().__init__(uri)

        self.entries: Dict[str, CatalogEntry] = {}