Skip to content

rdbs_storage

mlte/store/common/rdbs/store.py

Base RDBS store.

RDBStorage

Bases: Storage

Helper to setup RDB storage..

Source code in mlte/store/common/rdbs_storage.py 
20
21
22
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
class RDBStorage(Storage):
    """Helper to setup RDB storage.."""

    def __init__(
        self,
        uri: StoreURI,
        base_class: DeclarativeBase,
        init_tables_func: Optional[Callable[[sqlalchemy.Engine], None]],
        **kwargs,
    ) -> None:
        super().__init__(uri)

        self.engine = sqlalchemy.create_engine(uri.uri, **kwargs)
        """The underlying DB engine access."""

        # Create the DB if it doesn't exist already.
        if not sqlalchemy_utils.database_exists(self.engine.url):
            sqlalchemy_utils.create_database(self.engine.url)

        # Creates the DB items if they don't exist already.
        self._create_tables(base_class)
        if init_tables_func:
            init_tables_func(self.engine)

    def _create_tables(self, base_class: DeclarativeBase):
        """Creates all items, if they don't exist already."""
        base_class.metadata.create_all(self.engine)

    def close(self):
        """Cleanup."""
        self.engine.dispose()

engine = sqlalchemy.create_engine(uri.uri, **kwargs) instance-attribute

The underlying DB engine access.

close()

Cleanup.

Source code in mlte/store/common/rdbs_storage.py 
48
49
50
def close(self):
    """Cleanup."""
    self.engine.dispose()