Skip to content

memory

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 
 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
137
138
139
140
141
142
143
class InMemoryCatalogEntryMapper(CatalogEntryMapper):
    """In-memory mapper for the catalog resource."""

    def __init__(
        self, *, storage: MemoryCatalogStorage, validators: CompositeValidator
    ) -> None:
        super().__init__(validators=validators)

        self.storage = storage
        """A reference to underlying storage."""

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

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

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

        self.storage.entries[new_entry.header.identifier] = new_entry
        return new_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 
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
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,
            validators=self.validators,
            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 
44
45
46
47
48
49
50
51
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 
33
34
35
36
37
38
39
40
41
42
def session(self) -> CatalogStoreSession:
    """
    Return a session handle for the store instance.
    :return: The session handle
    """
    return InMemoryCatalogStoreSession(
        storage=self.storage,
        validators=self.validators,
        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 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
class InMemoryCatalogStoreSession(CatalogStoreSession):
    """An in-memory implementation of the MLTE user store."""

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

    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 
88
89
90
91
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 
54
55
56
57
58
59
60
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] = {}