Skip to content

store

Implementation of relation database system custom list store.

RDBCustomListEntryMapper

Bases: CustomListEntryMapper

RDB mapper mapper for the custom list entry resource.

Source code in mlte/store/custom_list/underlying/rdbs/store.py 
 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
137
138
139
140
141
142
143
144
145
146
class RDBCustomListEntryMapper(CustomListEntryMapper):
    """RDB mapper mapper for the custom list entry resource."""

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

    def create(
        self,
        new_entry: CustomListEntryModel,
        list_name: Optional[CustomListName] = None,
    ) -> CustomListEntryModel:
        list_name = self._check_valid_custom_list(list_name)
        self._ensure_parent_exists(new_entry.parent, list_name)

        with Session(self.storage.engine) as session:
            try:
                _, _ = DBReader.get_entry(new_entry.name, session)
                raise errors.ErrorAlreadyExists(
                    f"Custom list entry with name {new_entry.name} already exists."
                )
            except errors.ErrorNotFound:
                # If entry was not found, it means we can create it.
                entry_orm = DBReader.create_custom_list_entry_orm(
                    new_entry,
                    list_name,
                    session,
                )
                session.add(entry_orm)
                session.commit()
                return new_entry

    def read(
        self, entry_name: str, list_name: Optional[CustomListName] = None
    ) -> CustomListEntryModel:
        with Session(self.storage.engine) as session:
            entry, _ = DBReader.get_entry(entry_name, session)
            return entry

    def list(self, list_name: Optional[CustomListName] = None) -> list[str]:
        list_name = self._check_valid_custom_list(list_name)
        with Session(self.storage.engine) as session:
            entries, _ = DBReader.get_list(list_name, session)
            return [entry.name for entry in entries]

    def edit(
        self,
        updated_entry: CustomListEntryModel,
        list_name: Optional[CustomListName] = None,
    ) -> CustomListEntryModel:
        list_name = self._check_valid_custom_list(list_name)
        self._ensure_parent_exists(updated_entry.parent, list_name)

        with Session(self.storage.engine) as session:
            _, entry_orm = DBReader.get_entry(updated_entry.name, session)

            # Update existing entry
            entry_orm = DBReader.create_custom_list_entry_orm(
                updated_entry, entry_orm.list_name, session, entry_orm
            )
            session.commit()

            stored_entry, _ = DBReader.get_entry(updated_entry.name, session)
            return stored_entry

    def delete(
        self, entry_name: str, list_name: Optional[CustomListName] = None
    ) -> CustomListEntryModel:
        list_name = self._check_valid_custom_list(list_name)

        with Session(self.storage.engine) as session:
            entry, entry_orm = DBReader.get_entry(entry_name, session)
            session.delete(entry_orm)
            session.commit()
            return entry

storage = storage instance-attribute

A reference to underlying storage.

RDBCustomListStore

Bases: CustomListStore

A DB implementation of the MLTE custom list store.

Source code in mlte/store/custom_list/underlying/rdbs/store.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
class RDBCustomListStore(CustomListStore):
    """A DB implementation of the MLTE custom list store."""

    def __init__(self, uri: StoreURI, **kwargs) -> None:
        self.storage = RDBStorage(
            uri,
            base_class=typing.cast(DeclarativeBase, DBBase),
            **kwargs,
        )
        """The relational DB storage."""

    def session(self) -> RDBCustomListStoreSession:
        """
        Return a session handle fo the store session.
        :return: The session handle
        """
        return RDBCustomListStoreSession(storage=self.storage)

storage = RDBStorage(uri, base_class=(typing.cast(DeclarativeBase, DBBase)), **kwargs) instance-attribute

The relational DB storage.

session()

Return a session handle fo the store session.

Returns:

Type Description
RDBCustomListStoreSession

The session handle
Source code in mlte/store/custom_list/underlying/rdbs/store.py 
39
40
41
42
43
44
def session(self) -> RDBCustomListStoreSession:
    """
    Return a session handle fo the store session.
    :return: The session handle
    """
    return RDBCustomListStoreSession(storage=self.storage)

RDBCustomListStoreSession

Bases: CustomListStoreSession

A relational DB implementation of the MLTE custom list store session.

Source code in mlte/store/custom_list/underlying/rdbs/store.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
class RDBCustomListStoreSession(CustomListStoreSession):
    """A relational DB implementation of the MLTE custom list store session."""

    def __init__(self, storage: RDBStorage) -> None:
        self.storage = storage
        """"RDB storage."""

        self.custom_list_entry_mapper = RDBCustomListEntryMapper(storage)
        """The mapper to entry CRUD"""

    def close(self) -> None:
        """Close the session."""
        self.storage.close()

custom_list_entry_mapper = RDBCustomListEntryMapper(storage) instance-attribute

The mapper to entry CRUD

storage = storage instance-attribute

"RDB storage.

close()

Close the session.

Source code in mlte/store/custom_list/underlying/rdbs/store.py 
62
63
64
def close(self) -> None:
    """Close the session."""
    self.storage.close()