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 
 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
110
111
112
113
114
115
116
117
118
119
120
121
class UnifiedStore:
    """
    Unifies all store handling into one class.
    """

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

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

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

        self._user_store: UserStore | None = 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_store(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,
        )

    def import_store(self, store_data: dict[str, Any], force: bool = False):
        """Import store data"""
        import_store(
            store_data,
            self.artifact_store,
            self.custom_list_store,
            self.user_store,
            self.catalog_stores,
            force,
        )

    @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 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def __init__(self):
    """Defines the existing stores, none loaded yet."""

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

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

    self._user_store: UserStore | None = 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 
63
64
65
66
67
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 
69
70
71
72
73
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_store(export_spec, output_path)

Export store data.

Source code in mlte/store/unified_store.py 
75
76
77
78
79
80
81
82
83
84
def export_store(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,
    )

import_store(store_data, force=False)

Import store data

Source code in mlte/store/unified_store.py 
86
87
88
89
90
91
92
93
94
95
def import_store(self, store_data: dict[str, Any], force: bool = False):
    """Import store data"""
    import_store(
        store_data,
        self.artifact_store,
        self.custom_list_store,
        self.user_store,
        self.catalog_stores,
        force,
    )

set_artifact_store(store)

Set the artifact store.

Source code in mlte/store/unified_store.py 
51
52
53
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 
55
56
57
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 
59
60
61
def set_user_store(self, store: UserStore) -> None:
    """Set the user store."""
    self._user_store = store

setup_stores(stores_uri, catalog_uris=None)

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] | None

A dict of URIs for catalog stores.

 None
Source code in mlte/store/unified_store.py 
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
def setup_stores(
    stores_uri: StoreURI,
    catalog_uris: dict[str, StoreURI] | None = None,
) -> 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 if catalog_uris else {}
    )

    return stores