Skip to content

store

mlte/store/artifact/store.py

MLTE artifact store interface implementation.

ArtifactStore

Bases: Store

An abstract artifact store.

A Store instance is the "static" part of a store configuration. In contrast, a StoreSession represents an active session with the store.

Source code in mlte/store/artifact/store.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
class ArtifactStore(Store):
    """
    An abstract artifact store.

    A Store instance is the "static" part of a store configuration.
    In contrast, a StoreSession represents an active session with the store.
    """

    def session(self) -> ArtifactStoreSession:
        """
        Return a session handle for the store instance.
        :return: The session handle
        """
        raise NotImplementedError("Can't call session on a base Store.")

session()

Return a session handle for the store instance.

Returns:

Type Description
ArtifactStoreSession

The session handle
Source code in mlte/store/artifact/store.py 
30
31
32
33
34
35
def session(self) -> ArtifactStoreSession:
    """
    Return a session handle for the store instance.
    :return: The session handle
    """
    raise NotImplementedError("Can't call session on a base Store.")

ArtifactStoreSession

Bases: StoreSession

The base class for all implementations of the MLTE artifact store session.

Source code in mlte/store/artifact/store.py 
 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
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
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
class ArtifactStoreSession(StoreSession):
    """The base class for all implementations of the MLTE artifact store session."""

    # -------------------------------------------------------------------------
    # Interface: Context
    # -------------------------------------------------------------------------

    def create_model(self, model: Model) -> Model:
        """
        Create a MLTE model.
        :param model: The model data to create the model
        :return: The created model
        """
        raise NotImplementedError(
            "Cannot invoke method on abstract ArtifactStoreSession."
        )

    def read_model(self, model_id: str) -> Model:
        """
        Read a MLTE model.
        :param model_id: The identifier for the model
        :return: The model
        """
        raise NotImplementedError(
            "Cannot invoke method on abstract ArtifactStoreSession."
        )

    def list_models(self) -> List[str]:
        """
        List all MLTE models in the store.
        :return: A collection of identifiers for all MLTE models
        """
        raise NotImplementedError(
            "Cannot invoke method on abstract ArtifactStoreSession."
        )

    def delete_model(self, model_id: str) -> Model:
        """
        Delete a MLTE model.
        :param model_id: The identifier for the model
        :return: The deleted model
        """
        raise NotImplementedError(
            "Cannot invoke method on abstract ArtifactStoreSession."
        )

    def create_version(self, model_id: str, version: Version) -> Version:
        """
        Create a MLTE model version.
        :param model_id: The identifier for the model
        :param version: The version create model
        :return: The created version
        """
        raise NotImplementedError(
            "Cannot invoke method on abstract ArtifactStoreSession."
        )

    def list_versions(self, model_id: str) -> List[str]:
        """
        List all MLTE versions in the given model.
        :param model_id: The identifier for the model
        :return: A collection of identifiers for all MLTE versions
        """
        raise NotImplementedError(
            "Cannot invoke method on abstract ArtifactStoreSession."
        )

    def read_version(self, model_id: str, version_id: str) -> Version:
        """
        Read a MLTE model version.
        :param model_id: The identifier for the model
        :param version_id: The identifier for the model version
        :return: The version model
        """
        raise NotImplementedError(
            "Cannot invoke method on abstract ArtifactStoreSession."
        )

    def delete_version(self, model_id: str, version_id: str) -> Version:
        """
        Delete a MLTE model version.
        :param model_id: The identifier for the model
        :param version_id: The identifier for the model version
        :return: The deleted version
        """
        raise NotImplementedError(
            "Cannot invoke method on abstract ArtifactStoreSession."
        )

    # -------------------------------------------------------------------------
    # Interface: Artifact
    # -------------------------------------------------------------------------

    def write_artifact_with_header(
        self,
        model_id: str,
        version_id: str,
        artifact: ArtifactModel,
        *,
        force: bool = False,
        parents: bool = False,
        user: Optional[str] = None,
    ) -> ArtifactModel:
        """
        Write an artifact, generating the timestamp and adding creator. Internally calls the actual write_artifact implementation.
        :param model_id: The identifier for the model
        :param version_id: The identifier for the model version
        :param artifact: The artifact
        :param force: Overwrite an artifact if it already exists
        :param parents: Indicates whether organizational elements
        for artifact should be implictly created (default: False)
        """
        artifact.header.timestamp = int(time.time())
        artifact.header.creator = user
        return self.write_artifact(
            model_id,
            version_id,
            artifact,
            force=force,
            parents=parents,
        )

    def write_artifact(
        self,
        model_id: str,
        version_id: str,
        artifact: ArtifactModel,
        *,
        force: bool = False,
        parents: bool = False,
    ) -> ArtifactModel:
        """
        Write an artifact.
        :param model_id: The identifier for the model
        :param version_id: The identifier for the model version
        :param artifact: The artifact
        :param force: Overwrite an artifact if it already exists
        :param parents: Indicates whether organizational elements
        for artifact should be implictly created (default: False)
        """
        raise NotImplementedError(
            "Cannot invoke method on abstract ArtifactStoreSession."
        )

    def read_artifact(
        self,
        model_id: str,
        version_id: str,
        artifact_id: str,
    ) -> ArtifactModel:
        """
        Read an artifact.
        :param model_id: The identifier for the model
        :param version_id: The identifier for the model version
        :param artifact_id: The artifact identifier
        :return: The artifact
        """
        raise NotImplementedError(
            "Cannot invoke method on abstract ArtifactStoreSession."
        )

    def read_artifacts(
        self,
        model_id: str,
        version_id: str,
        limit: int = 100,
        offset: int = 0,
    ) -> List[ArtifactModel]:
        """
        Read artifacts withi limit and offset.
        :param model_id: The identifier for the model
        :param version_id: The identifier for the model version
        :param limit: The limit on artifacts to read
        :param offset: The offset on artifacts to read
        :return: The read artifacts
        """
        raise NotImplementedError(
            "Cannot invoke method on abstract ArtifactStoreSession."
        )

    def search_artifacts(
        self,
        model_id: str,
        version_id: str,
        query: Query = Query(),
    ) -> List[ArtifactModel]:
        """
        Read a collection of artifacts, optionally filtered.
        :param model_id: The identifier for the model
        :param version_id: The identifier for the model version
        :param query: The artifact query to apply
        :return: A collection of artifacts that satisfy the filter
        """
        raise NotImplementedError(
            "Cannot invoke method on abstract ArtifactStoreSession."
        )

    def delete_artifact(
        self,
        model_id: str,
        version_id: str,
        artifact_id: str,
    ) -> ArtifactModel:
        """
        Delete an artifact.
        :param model_id: The identifier for the model
        :param version_id: The identifier for the model version
        :param artifact_id: The artifact identifier
        :return: The deleted artifact
        """
        raise NotImplementedError(
            "Cannot invoke method on abstract ArtifactStoreSession."
        )

create_model(model)

Create a MLTE model.

Parameters:

Name Type Description Default
model Model

The model data to create the model

 required

Returns:

Type Description
Model

The created model
Source code in mlte/store/artifact/store.py 
50
51
52
53
54
55
56
57
58
def create_model(self, model: Model) -> Model:
    """
    Create a MLTE model.
    :param model: The model data to create the model
    :return: The created model
    """
    raise NotImplementedError(
        "Cannot invoke method on abstract ArtifactStoreSession."
    )

create_version(model_id, version)

Create a MLTE model version.

Parameters:

Name Type Description Default
model_id str

The identifier for the model

 required
version Version

The version create model

 required

Returns:

Type Description
Version

The created version
Source code in mlte/store/artifact/store.py 
89
90
91
92
93
94
95
96
97
98
def create_version(self, model_id: str, version: Version) -> Version:
    """
    Create a MLTE model version.
    :param model_id: The identifier for the model
    :param version: The version create model
    :return: The created version
    """
    raise NotImplementedError(
        "Cannot invoke method on abstract ArtifactStoreSession."
    )

delete_artifact(model_id, version_id, artifact_id)

Delete an artifact.

Parameters:

Name Type Description Default
model_id str

The identifier for the model

 required
version_id str

The identifier for the model version

 required
artifact_id str

The artifact identifier

 required

Returns:

Type Description
ArtifactModel

The deleted artifact
Source code in mlte/store/artifact/store.py 
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
def delete_artifact(
    self,
    model_id: str,
    version_id: str,
    artifact_id: str,
) -> ArtifactModel:
    """
    Delete an artifact.
    :param model_id: The identifier for the model
    :param version_id: The identifier for the model version
    :param artifact_id: The artifact identifier
    :return: The deleted artifact
    """
    raise NotImplementedError(
        "Cannot invoke method on abstract ArtifactStoreSession."
    )

delete_model(model_id)

Delete a MLTE model.

Parameters:

Name Type Description Default
model_id str

The identifier for the model

 required

Returns:

Type Description
Model

The deleted model
Source code in mlte/store/artifact/store.py 
79
80
81
82
83
84
85
86
87
def delete_model(self, model_id: str) -> Model:
    """
    Delete a MLTE model.
    :param model_id: The identifier for the model
    :return: The deleted model
    """
    raise NotImplementedError(
        "Cannot invoke method on abstract ArtifactStoreSession."
    )

delete_version(model_id, version_id)

Delete a MLTE model version.

Parameters:

Name Type Description Default
model_id str

The identifier for the model

 required
version_id str

The identifier for the model version

 required

Returns:

Type Description
Version

The deleted version
Source code in mlte/store/artifact/store.py 
121
122
123
124
125
126
127
128
129
130
def delete_version(self, model_id: str, version_id: str) -> Version:
    """
    Delete a MLTE model version.
    :param model_id: The identifier for the model
    :param version_id: The identifier for the model version
    :return: The deleted version
    """
    raise NotImplementedError(
        "Cannot invoke method on abstract ArtifactStoreSession."
    )

list_models()

List all MLTE models in the store.

Returns:

Type Description
List[str]

A collection of identifiers for all MLTE models
Source code in mlte/store/artifact/store.py 
70
71
72
73
74
75
76
77
def list_models(self) -> List[str]:
    """
    List all MLTE models in the store.
    :return: A collection of identifiers for all MLTE models
    """
    raise NotImplementedError(
        "Cannot invoke method on abstract ArtifactStoreSession."
    )

list_versions(model_id)

List all MLTE versions in the given model.

Parameters:

Name Type Description Default
model_id str

The identifier for the model

 required

Returns:

Type Description
List[str]

A collection of identifiers for all MLTE versions
Source code in mlte/store/artifact/store.py 
100
101
102
103
104
105
106
107
108
def list_versions(self, model_id: str) -> List[str]:
    """
    List all MLTE versions in the given model.
    :param model_id: The identifier for the model
    :return: A collection of identifiers for all MLTE versions
    """
    raise NotImplementedError(
        "Cannot invoke method on abstract ArtifactStoreSession."
    )

read_artifact(model_id, version_id, artifact_id)

Read an artifact.

Parameters:

Name Type Description Default
model_id str

The identifier for the model

 required
version_id str

The identifier for the model version

 required
artifact_id str

The artifact identifier

 required

Returns:

Type Description
ArtifactModel

The artifact
Source code in mlte/store/artifact/store.py 
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
def read_artifact(
    self,
    model_id: str,
    version_id: str,
    artifact_id: str,
) -> ArtifactModel:
    """
    Read an artifact.
    :param model_id: The identifier for the model
    :param version_id: The identifier for the model version
    :param artifact_id: The artifact identifier
    :return: The artifact
    """
    raise NotImplementedError(
        "Cannot invoke method on abstract ArtifactStoreSession."
    )

read_artifacts(model_id, version_id, limit=100, offset=0)

Read artifacts withi limit and offset.

Parameters:

Name Type Description Default
model_id str

The identifier for the model

 required
version_id str

The identifier for the model version

 required
limit int

The limit on artifacts to read

 100
offset int

The offset on artifacts to read

 0

Returns:

Type Description
List[ArtifactModel]

The read artifacts
Source code in mlte/store/artifact/store.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
def read_artifacts(
    self,
    model_id: str,
    version_id: str,
    limit: int = 100,
    offset: int = 0,
) -> List[ArtifactModel]:
    """
    Read artifacts withi limit and offset.
    :param model_id: The identifier for the model
    :param version_id: The identifier for the model version
    :param limit: The limit on artifacts to read
    :param offset: The offset on artifacts to read
    :return: The read artifacts
    """
    raise NotImplementedError(
        "Cannot invoke method on abstract ArtifactStoreSession."
    )

read_model(model_id)

Read a MLTE model.

Parameters:

Name Type Description Default
model_id str

The identifier for the model

 required

Returns:

Type Description
Model

The model
Source code in mlte/store/artifact/store.py 
60
61
62
63
64
65
66
67
68
def read_model(self, model_id: str) -> Model:
    """
    Read a MLTE model.
    :param model_id: The identifier for the model
    :return: The model
    """
    raise NotImplementedError(
        "Cannot invoke method on abstract ArtifactStoreSession."
    )

read_version(model_id, version_id)

Read a MLTE model version.

Parameters:

Name Type Description Default
model_id str

The identifier for the model

 required
version_id str

The identifier for the model version

 required

Returns:

Type Description
Version

The version model
Source code in mlte/store/artifact/store.py 
110
111
112
113
114
115
116
117
118
119
def read_version(self, model_id: str, version_id: str) -> Version:
    """
    Read a MLTE model version.
    :param model_id: The identifier for the model
    :param version_id: The identifier for the model version
    :return: The version model
    """
    raise NotImplementedError(
        "Cannot invoke method on abstract ArtifactStoreSession."
    )

search_artifacts(model_id, version_id, query=Query())

Read a collection of artifacts, optionally filtered.

Parameters:

Name Type Description Default
model_id str

The identifier for the model

 required
version_id str

The identifier for the model version

 required
query Query

The artifact query to apply

 Query()

Returns:

Type Description
List[ArtifactModel]

A collection of artifacts that satisfy the filter
Source code in mlte/store/artifact/store.py 
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
def search_artifacts(
    self,
    model_id: str,
    version_id: str,
    query: Query = Query(),
) -> List[ArtifactModel]:
    """
    Read a collection of artifacts, optionally filtered.
    :param model_id: The identifier for the model
    :param version_id: The identifier for the model version
    :param query: The artifact query to apply
    :return: A collection of artifacts that satisfy the filter
    """
    raise NotImplementedError(
        "Cannot invoke method on abstract ArtifactStoreSession."
    )

write_artifact(model_id, version_id, artifact, *, force=False, parents=False)

Write an artifact.

Parameters:

Name Type Description Default
model_id str

The identifier for the model

 required
version_id str

The identifier for the model version

 required
artifact ArtifactModel

The artifact

 required
force bool

Overwrite an artifact if it already exists

 False
parents bool

Indicates whether organizational elements for artifact should be implictly created (default: False)

 False
Source code in mlte/store/artifact/store.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
def write_artifact(
    self,
    model_id: str,
    version_id: str,
    artifact: ArtifactModel,
    *,
    force: bool = False,
    parents: bool = False,
) -> ArtifactModel:
    """
    Write an artifact.
    :param model_id: The identifier for the model
    :param version_id: The identifier for the model version
    :param artifact: The artifact
    :param force: Overwrite an artifact if it already exists
    :param parents: Indicates whether organizational elements
    for artifact should be implictly created (default: False)
    """
    raise NotImplementedError(
        "Cannot invoke method on abstract ArtifactStoreSession."
    )

write_artifact_with_header(model_id, version_id, artifact, *, force=False, parents=False, user=None)

Write an artifact, generating the timestamp and adding creator. Internally calls the actual write_artifact implementation.

Parameters:

Name Type Description Default
model_id str

The identifier for the model

 required
version_id str

The identifier for the model version

 required
artifact ArtifactModel

The artifact

 required
force bool

Overwrite an artifact if it already exists

 False
parents bool

Indicates whether organizational elements for artifact should be implictly created (default: False)

 False
Source code in mlte/store/artifact/store.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def write_artifact_with_header(
    self,
    model_id: str,
    version_id: str,
    artifact: ArtifactModel,
    *,
    force: bool = False,
    parents: bool = False,
    user: Optional[str] = None,
) -> ArtifactModel:
    """
    Write an artifact, generating the timestamp and adding creator. Internally calls the actual write_artifact implementation.
    :param model_id: The identifier for the model
    :param version_id: The identifier for the model version
    :param artifact: The artifact
    :param force: Overwrite an artifact if it already exists
    :param parents: Indicates whether organizational elements
    for artifact should be implictly created (default: False)
    """
    artifact.header.timestamp = int(time.time())
    artifact.header.creator = user
    return self.write_artifact(
        model_id,
        version_id,
        artifact,
        force=force,
        parents=parents,
    )

ManagedArtifactSession

Bases: ManagedSession

A simple context manager for store sessions.

Source code in mlte/store/artifact/store.py 
258
259
260
261
262
class ManagedArtifactSession(ManagedSession):
    """A simple context manager for store sessions."""

    def __enter__(self) -> ArtifactStoreSession:
        return cast(ArtifactStoreSession, self.session)