Skip to content

store

mlte/store/catalog/underlying/rdbs/store.py

Implementation of relational database system catalog store.

RDBEntryMapper

Bases: CatalogEntryMapper

RDB mapper for a catalog entry resource.

Source code in mlte/store/catalog/underlying/rdbs/store.py 
 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
class RDBEntryMapper(CatalogEntryMapper):
    """RDB mapper for a catalog entry resource."""

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

    def create(self, entry: CatalogEntry, context: Any = None) -> CatalogEntry:
        with Session(self.storage.engine) as session:
            try:
                _, _ = DBReader.get_entry(entry.header.identifier, session)
                raise errors.ErrorAlreadyExists(
                    f"Entry with id {entry.header.identifier} already exists."
                )
            except errors.ErrorNotFound:
                # If it was not found, it means we can create it.
                entry_obj = DBReader._build_entry_obj(entry, session)
                session.add(entry_obj)
                session.commit()

                stored_entry, _ = DBReader.get_entry(
                    entry.header.identifier, session
                )
                return stored_entry

    def edit(self, entry: CatalogEntry, context: Any = None) -> CatalogEntry:
        with Session(self.storage.engine) as session:
            _, entry_obj = DBReader.get_entry(entry.header.identifier, session)

            # Update existing user.
            entry_obj = DBReader._build_entry_obj(entry, session, entry_obj)
            session.commit()

            stored_entry, _ = DBReader.get_entry(
                entry.header.identifier, session
            )
            return stored_entry

    def read(self, entry_id: str, context: Any = None) -> CatalogEntry:
        with Session(self.storage.engine) as session:
            catalog_entry, _ = DBReader.get_entry(entry_id, session)
            return catalog_entry

    def list(self, context: Any = None) -> List[str]:
        with Session(self.storage.engine) as session:
            entries, _ = DBReader.get_entries(session)
            return [entry.header.identifier for entry in entries]

    def delete(self, entry_id: str, context: Any = None) -> CatalogEntry:
        with Session(self.storage.engine) as session:
            catalog_entry, entry_obj = DBReader.get_entry(entry_id, session)
            session.delete(entry_obj)
            session.commit()
            return catalog_entry

storage = storage instance-attribute

A reference to underlying storage.

RelationalDBCatalogStore

Bases: CatalogStore

A DB implementation of the MLTE user store.

Source code in mlte/store/catalog/underlying/rdbs/store.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
class RelationalDBCatalogStore(CatalogStore):
    """A DB implementation of the MLTE user store."""

    def __init__(self, uri, **kwargs):
        CatalogStore.__init__(self, uri=uri)

        self.storage = RDBStorage(
            uri,
            base_class=typing.cast(DeclarativeBase, DBBase),
            init_tables_func=init_catalog_tables,
            **kwargs,
        )
        """The relational DB storage."""

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

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

The relational DB storage.

session()

Return a session handle for the store instance.

Returns:

Type Description
RelationalDBCatalogStoreSession

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

RelationalDBCatalogStoreSession

Bases: CatalogStoreSession

A relational DB implementation of the MLTE user store session.

Source code in mlte/store/catalog/underlying/rdbs/store.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
class RelationalDBCatalogStoreSession(CatalogStoreSession):
    """A relational DB implementation of the MLTE user store session."""

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

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

        self.entry_mapper = RDBEntryMapper(storage)
        """The mapper to user CRUD."""

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

entry_mapper = RDBEntryMapper(storage) instance-attribute

The mapper to user CRUD.

read_only = read_only instance-attribute

Whether this is read only or not.

storage = storage instance-attribute

RDB storage.

close()

Close the session.

Source code in mlte/store/catalog/underlying/rdbs/store.py 
82
83
84
def close(self) -> None:
    """Close the session."""
    self.storage.close()

init_catalog_tables(engine)

Pre-populate tables.

Source code in mlte/store/catalog/underlying/rdbs/store.py 
58
59
60
61
def init_catalog_tables(engine: Engine):
    """Pre-populate tables."""
    with Session(engine) as session:
        init_catalog_entry_types(session)