Skip to content

memory

Implementation of in-memory custom list store.

InMemoryCustomListEntryMapper

Bases: CustomListEntryMapper

In-memory mapper for custom list entry resource

Source code in mlte/store/custom_list/underlying/memory.py 
 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
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
class InMemoryCustomListEntryMapper(CustomListEntryMapper):
    """In-memory mapper for custom list entry resource"""

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

    def create(
        self,
        entry: CustomListEntryModel,
        list_name: Optional[CustomListName] = None,
    ) -> CustomListEntryModel:
        list_name = self._check_valid_custom_list(list_name)
        self._ensure_parent_exists(entry.parent, list_name)
        if entry.name in self.storage.custom_lists[list_name]:
            raise errors.ErrorAlreadyExists(f"Custom list Entry {entry.name}")

        self.storage.custom_lists[list_name][entry.name] = entry
        return entry

    def read(
        self, entry_name: str, list_name: Optional[CustomListName] = None
    ) -> CustomListEntryModel:
        list_name = self._check_valid_custom_list(list_name)
        self._check_entry_in_list(entry_name, list_name)
        entry = self.storage.custom_lists[list_name][entry_name]
        return entry

    def list(self, list_name: Optional[CustomListName] = None) -> List[str]:
        list_name = self._check_valid_custom_list(list_name)
        return [
            entry_name
            for entry_name in self.storage.custom_lists[list_name].keys()
        ]

    def edit(
        self,
        entry: CustomListEntryModel,
        list_name: Optional[CustomListName] = None,
    ) -> CustomListEntryModel:
        list_name = self._check_valid_custom_list(list_name)
        self._ensure_parent_exists(entry.parent, list_name)
        self._check_entry_in_list(entry.name, list_name)
        self.storage.custom_lists[list_name][entry.name] = entry
        return entry

    def delete(
        self, entry_name: str, list_name: Optional[CustomListName] = None
    ) -> CustomListEntryModel:
        list_name = self._check_valid_custom_list(list_name)
        self._check_entry_in_list(entry_name, list_name)
        self._delete_children(list_name, entry_name)
        popped = self.storage.custom_lists[list_name][entry_name]
        del self.storage.custom_lists[list_name][entry_name]
        return popped

    def _check_valid_custom_list(
        self, list_name: Optional[CustomListName]
    ) -> CustomListName:
        """Checks if the custom lists exists within the store."""
        if list_name is None or list_name not in self.storage.custom_lists:
            raise errors.ErrorNotFound(
                f"CustomListName, {list_name}, does not exist or is None."
            )
        else:
            return list_name

    def _check_entry_in_list(self, entry_name: str, list_name: CustomListName):
        if entry_name not in self.storage.custom_lists[list_name]:
            raise errors.ErrorNotFound(f"Custom list Entry {entry_name}")

storage = storage instance-attribute

A reference to underlying storage.

InMemoryCustomListStore

Bases: CustomListStore

An in-memory implementation of the MLTE custom list store.

Source code in mlte/store/custom_list/underlying/memory.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
class InMemoryCustomListStore(CustomListStore):
    """An in-memory implementation of the MLTE custom list store."""

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

        super().__init__(uri=uri)

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

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

storage = MemoryCustomListStorage() instance-attribute

The underlying storage for the store.

clone()

Clones the store. Shallow clone.

Returns:

Type Description
InMemoryCustomListStore

The cloned store
Source code in mlte/store/custom_list/underlying/memory.py 
38
39
40
41
42
43
44
45
def clone(self) -> InMemoryCustomListStore:
    """
    Clones the store. Shallow clone.
    :return: The cloned store
    """
    clone = InMemoryCustomListStore(self.uri)
    clone.storage.custom_lists = self.storage.custom_lists.copy()
    return clone

session()

Return a session handle for the store instance.

Returns:

Type Description
InMemoryCustomListStoreSession

The session handle
Source code in mlte/store/custom_list/underlying/memory.py 
31
32
33
34
35
36
def session(self) -> InMemoryCustomListStoreSession:
    """
    Return a session handle for the store instance.
    :return: The session handle
    """
    return InMemoryCustomListStoreSession(storage=self.storage)

InMemoryCustomListStoreSession

Bases: CustomListStoreSession

An in-memory implementation of the MLTE custom list store.

Source code in mlte/store/custom_list/underlying/memory.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
class InMemoryCustomListStoreSession(CustomListStoreSession):
    """An in-memory implementation of the MLTE custom list store."""

    def __init__(self, *, storage: MemoryCustomListStorage) -> None:
        self.storage = storage
        """The storage."""

        self.custom_list_entry_mapper = InMemoryCustomListEntryMapper(
            storage=storage
        )
        """The mapper to the custom list entry CRUD."""

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

custom_list_entry_mapper = InMemoryCustomListEntryMapper(storage=storage) instance-attribute

The mapper to the custom list entry CRUD.

storage = storage instance-attribute

The storage.

close()

Close the session.

Source code in mlte/store/custom_list/underlying/memory.py 
76
77
78
79
def close(self) -> None:
    """Close the session."""
    # Closing an in-memory session is a no-op.
    pass

MemoryCustomListStorage

A simple storage wrapper for the in-memory store.

Source code in mlte/store/custom_list/underlying/memory.py 
48
49
50
51
52
53
54
55
56
class MemoryCustomListStorage:
    """A simple storage wrapper for the in-memory store."""

    def __init__(self) -> None:
        self.custom_lists: Dict[
            CustomListName, Dict[str, CustomListEntryModel]
        ] = {}
        for list_name in CustomListName:
            self.custom_lists[list_name] = {}