Skip to content

unified_store

Manages info about a unified set of stores.

UnifiedStore

Unifies all store handling into one class.

Source code in mlte/store/unified_store.py 
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 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
class UnifiedStore:
    """
    Unifies all store handling into one class.
    """

    def __init__(self):
        """Defines the existing stores, none loaded yet."""

        self._artifact_store: Optional[ArtifactStore] = None
        """The MLTE artifact store instance for the session."""

        self._custom_list_store: Optional[CustomListStore] = None
        """The MLTE custom list store instance for the session."""

        self._user_store: Optional[UserStore] = None
        """The user store instance for the session."""

        self._catalog_stores: CatalogStoreGroup = CatalogStoreGroup()
        """The list of catalog store instances maintained by the session object."""

    def set_artifact_store(self, store: ArtifactStore) -> None:
        """Set the artifact store."""
        self._artifact_store = store

    def set_custom_list_store(self, store: CustomListStore) -> None:
        """Set the custom list store."""
        self._custom_list_store = store

    def set_user_store(self, store: UserStore) -> None:
        """Set the user store."""
        self._user_store = store

    def add_catalog_store(
        self, store: CatalogStore, id: str, overwite: bool = False
    ) -> None:
        """Adds to the the list of catalog stores."""
        self._catalog_stores.add_catalog(id, store, overwite)

    def add_catalog_store_from_uri(
        self, store_uri: StoreURI, id: str, overwite: bool = False
    ) -> None:
        """Adds to the the list of catalog stores."""
        self._catalog_stores.add_catalog_from_uri(id, store_uri, overwite)

    def export(self, export_spec: ExportSpec, output_path: Path):
        """Export store data."""
        export_to_file(
            export_spec,
            output_path,
            self.artifact_store,
            self.custom_list_store,
            self.user_store,
            self.catalog_stores,
        )

    @property
    def artifact_store(self) -> ArtifactStore:
        """Get the session artifact store."""
        if not self._artifact_store:
            raise RuntimeError("Store was not properly set up.")
        return self._artifact_store

    @property
    def custom_list_store(self) -> CustomListStore:
        """Get the session custom list store."""
        if not self._custom_list_store:
            raise RuntimeError("Store was not properly set up.")
        return self._custom_list_store

    @property
    def user_store(self) -> UserStore:
        """Get session user store."""
        if not self._user_store:
            raise RuntimeError("Store was not properly set up.")
        return self._user_store

    @property
    def catalog_stores(self) -> CatalogStoreGroup:
        """Get all catalog stores."""
        return self._catalog_stores

artifact_store property

Get the session artifact store.

catalog_stores property

Get all catalog stores.

custom_list_store property

Get the session custom list store.

user_store property

Get session user store.

__init__()

Defines the existing stores, none loaded yet.

Source code in mlte/store/unified_store.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def __init__(self):
    """Defines the existing stores, none loaded yet."""

    self._artifact_store: Optional[ArtifactStore] = None
    """The MLTE artifact store instance for the session."""

    self._custom_list_store: Optional[CustomListStore] = None
    """The MLTE custom list store instance for the session."""

    self._user_store: Optional[UserStore] = None
    """The user store instance for the session."""

    self._catalog_stores: CatalogStoreGroup = CatalogStoreGroup()
    """The list of catalog store instances maintained by the session object."""

add_catalog_store(store, id, overwite=False)

Adds to the the list of catalog stores.

Source code in mlte/store/unified_store.py 
62
63
64
65
66
def add_catalog_store(
    self, store: CatalogStore, id: str, overwite: bool = False
) -> None:
    """Adds to the the list of catalog stores."""
    self._catalog_stores.add_catalog(id, store, overwite)

add_catalog_store_from_uri(store_uri, id, overwite=False)

Adds to the the list of catalog stores.

Source code in mlte/store/unified_store.py 
68
69
70
71
72
def add_catalog_store_from_uri(
    self, store_uri: StoreURI, id: str, overwite: bool = False
) -> None:
    """Adds to the the list of catalog stores."""
    self._catalog_stores.add_catalog_from_uri(id, store_uri, overwite)

export(export_spec, output_path)

Export store data.

Source code in mlte/store/unified_store.py 
74
75
76
77
78
79
80
81
82
83
def export(self, export_spec: ExportSpec, output_path: Path):
    """Export store data."""
    export_to_file(
        export_spec,
        output_path,
        self.artifact_store,
        self.custom_list_store,
        self.user_store,
        self.catalog_stores,
    )

set_artifact_store(store)

Set the artifact store.

Source code in mlte/store/unified_store.py 
50
51
52
def set_artifact_store(self, store: ArtifactStore) -> None:
    """Set the artifact store."""
    self._artifact_store = store

set_custom_list_store(store)

Set the custom list store.

Source code in mlte/store/unified_store.py 
54
55
56
def set_custom_list_store(self, store: CustomListStore) -> None:
    """Set the custom list store."""
    self._custom_list_store = store

set_user_store(store)

Set the user store.

Source code in mlte/store/unified_store.py 
58
59
60
def set_user_store(self, store: UserStore) -> None:
    """Set the user store."""
    self._user_store = store

setup_stores(stores_uri, catalog_uris={})

Sets up all stores required by MLTE, from the provided URIs.

Parameters:

Name Type Description Default
stores_uri StoreURI

The store URI string, used as the common type and root location for all non-catalog stores.

 required
catalog_uris dict[str, StoreURI]

A dict of URIs for catalog stores.

 {}
Source code in mlte/store/unified_store.py 
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
def setup_stores(
    stores_uri: StoreURI,
    catalog_uris: dict[str, StoreURI] = {},
) -> UnifiedStore:
    """
    Sets up all stores required by MLTE, from the provided URIs.

    :param stores_uri: The store URI string, used as the common type and root location for all non-catalog stores.
    :param catalog_uris: A dict of URIs for catalog stores.
    """
    stores = UnifiedStore()

    # Initialize the backing user store instance.
    user_store = user_store_factory.create_user_store(stores_uri)
    stores.set_user_store(user_store)

    # Initialize the backing custom list store instance.
    custom_list_store = InitialCustomLists.setup_custom_list_store(stores_uri)
    stores.set_custom_list_store(custom_list_store)

    # Initialize the backing artifact store instance.
    _setup_arifact_store(stores_uri, stores)

    # Initialize the backing catalog stores instances.
    _setup_catalog_stores(stores_uri, stores, catalog_uris)

    return stores