Skip to content

artifact

Artifact protocol implementation.

DEFAULT_ID = 'default' module-attribute

Default id used if none is provided. Full id will be prefixed by type.

Artifact

Bases: Serializable, ABC

The MLTE artifact protocol implementation.

The Artifact type establishes the common interface for all MLTE artifacts. This ensures that, even though they have very different semantics, all artifacts abide by a common protocol that allows us to perform common operations with them, namely persistence.

Source code in mlte/artifact/artifact.py 
 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
 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
256
257
258
259
260
261
class Artifact(Serializable, abc.ABC):
    """
    The MLTE artifact protocol implementation.

    The Artifact type establishes the common interface
    for all MLTE artifacts. This ensures that, even though
    they have very different semantics, all artifacts abide
    by a common protocol that allows us to perform common
    operations with them, namely persistence.
    """

    type: Optional[ArtifactType] = None
    """By default have no type, but a base Artifact should never be instantiated."""

    def __init__(self, identifier: Optional[str] = None) -> None:
        """Main constructor for all artifacts."""

        self.identifier = self.build_full_id(identifier)
        """
        The identifier for the artifact, always having its type as prefix.
        An artifact identifier is unique within a MLTE context
        (model, version) and for a given artifact type.
        """

        self.timestamp = -1
        """The Unix timestamp of when the artifact was saved to a store."""

        self.creator = None
        """The user that created this artifact."""

        self.level = ArtifactLevel.VERSION
        """The artifact level, be it model or version, defaults to version."""

    def __eq__(self, other: object) -> bool:
        """Test instance for equality."""
        if not isinstance(other, Artifact):
            return False
        return self._equal(other)

    def pre_save_hook(self, context: Context, store: ArtifactStore) -> None:
        """
        A method that artifact subclasses can override to enforce pre-save invariants.
        :param context: The context in which to save the artifact
        :param store: The store in which to save the artifact
        :raises RuntimeError: On broken invariant
        """
        # Default implementation is a no-op
        pass

    def post_load_hook(self, context: Context, store: ArtifactStore) -> None:
        """
        A method that artifact subclasses may override to enforce post-load invariants.
        :param context: The context in which to save the artifact
        :param store: The store in which to save the artifact
        :raises RuntimeError: On broken invariant
        """
        # Default implementation is a no-op
        pass

    def save(
        self,
        *,
        force: bool = False,
        parents: bool = False,
        user: Optional[str] = None,
    ) -> ArtifactModel:
        """
        Save an artifact with parameters from the configured global session.

        This is equivalent to calling:
            artifact.save_with(get_session().context, get_session().store)

        :param force: Indicates that an existing artifact may be overwritten
        :param parents: Indicates whether organizational elements for the
        artifact are created implicitly on write (default: False)
        :param user: The username of the user executing this action.
        :return: The ArtifactModel of the saved artifact.
        """
        credentials = get_session().credentials
        return self.save_with(
            get_session().context,
            get_session().stores.artifact_store,
            force=force,
            parents=parents,
            user=(
                user if user else (credentials.user if credentials else None)
            ),
        )

    def save_with(
        self,
        context: Context,
        store: ArtifactStore,
        *,
        force: bool = False,
        parents: bool = False,
        user: Optional[str] = None,
    ) -> ArtifactModel:
        """
        Save an artifact with the given context and store configuration.
        :param context: The context in which to save the artifact
        :param store: The store in which to save the artifact
        :param force: Indicates that an existing artifact may be overwritten
        :param parents: Indicates whether organizational elements for the
        artifact are created implicitly on write (default: False)
        :param user: The username of the user executing this action.
        :return: The ArtifactModel of the saved artifact.
        """
        with ManagedArtifactSession(store.session()) as artifact_store:
            # If we are forcing parent creation, ensure they are there before any hooks.
            if parents:
                artifact_store.create_parents(context.model, context.version)

            # Run any artifact-type specific pre save hooks.
            self.pre_save_hook(context, store)

            # Convert to model and save.
            model = self.to_model()

            assert isinstance(
                model, ArtifactModel
            ), "Can't create object from non-ArtifactModel model."

            return artifact_store.artifact_mapper.write_artifact_with_header(
                context.model,
                context.version,
                model,
                force=force,
                user=user,
            )

    @classmethod
    def load(cls, identifier: Optional[str] = None) -> Artifact:
        """
        Load an artifact from the configured global session.
        :param identifier: The identifier for the artifact. If None,
        the default id is used.

        This is equivalent to calling:
            Artifact.load_with(session().context, session().store)
        """
        return cls.load_with(
            identifier,
            context=get_session().context,
            store=get_session().stores.artifact_store,
        )

    @classmethod
    def load_with(
        cls,
        identifier: Optional[str] = None,
        *,
        context: Context,
        store: ArtifactStore,
    ) -> Artifact:
        """
        Load an artifact with the given context and store configuration.
        :param identifier: The identifier for the artifact If None,
        the default id is used.
        :param context: The context from which to load the artifact
        :param store: The store from which to load the artifact
        """
        identifier = cls.build_full_id(identifier)

        with ManagedArtifactSession(store.session()) as artifact_store:
            artifact = typing.cast(
                Artifact,
                cls.from_model(
                    artifact_store.artifact_mapper.read(
                        identifier, (context.model, context.version)
                    )
                ),
            )

        artifact.post_load_hook(context, store)
        return artifact

    @staticmethod
    def load_models_for_session(
        artifact_type: ArtifactType,
    ) -> list[ArtifactModel]:
        """Loads all artifact models of the given type from the session."""
        return Artifact.load_models(
            artifact_type,
            context=get_session().context,
            store=get_session().stores.artifact_store,
        )

    @staticmethod
    def load_models(
        artifact_type: ArtifactType, context: Context, store: ArtifactStore
    ) -> list[ArtifactModel]:
        """Loads all artifact models of the given type for the given context and store."""
        with ManagedArtifactSession(store.session()) as artifact_store:
            query_instance = Query(filter=TypeFilter(item_type=artifact_type))
            artifact_models = artifact_store.artifact_mapper.search(
                query_instance, context=(context.model, context.version)
            )
            return artifact_models

    @classmethod
    def build_full_id(cls, base: Optional[str] = None) -> str:
        """Builds the full id for this artifact. If base is None, default base is used."""
        if not cls.type:
            raise RuntimeError(
                "Malformed artifact class, type has not been set."
            )
        if not base:
            base = DEFAULT_ID
        return Artifact._build_id(cls.type.value, base)

    @staticmethod
    def _build_id(prefix: str, base: str) -> str:
        """Builds the common id structure for an artifact."""
        if not base.startswith(f"{prefix}."):
            return f"{prefix}.{base}"
        else:
            return base

    def build_artifact_header(self) -> ArtifactHeaderModel:
        """Generates the common header model for artifacts."""
        if not self.type:
            raise RuntimeError(
                "Malformed artifact class, type has not been set."
            )
        return ArtifactHeaderModel(
            identifier=self.identifier,
            type=self.type,
            timestamp=self.timestamp,
            creator=self.creator,
            level=self.level,
        )

    def __str__(self) -> str:
        """Return a string representation."""
        return self.to_model().to_json_string()

creator = None instance-attribute

The user that created this artifact.

identifier = self.build_full_id(identifier) instance-attribute

The identifier for the artifact, always having its type as prefix. An artifact identifier is unique within a MLTE context (model, version) and for a given artifact type.

level = ArtifactLevel.VERSION instance-attribute

The artifact level, be it model or version, defaults to version.

timestamp = -1 instance-attribute

The Unix timestamp of when the artifact was saved to a store.

type = None class-attribute instance-attribute

By default have no type, but a base Artifact should never be instantiated.

__eq__(other)

Test instance for equality.

Source code in mlte/artifact/artifact.py 
59
60
61
62
63
def __eq__(self, other: object) -> bool:
    """Test instance for equality."""
    if not isinstance(other, Artifact):
        return False
    return self._equal(other)

__init__(identifier=None)

Main constructor for all artifacts.

Source code in mlte/artifact/artifact.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def __init__(self, identifier: Optional[str] = None) -> None:
    """Main constructor for all artifacts."""

    self.identifier = self.build_full_id(identifier)
    """
    The identifier for the artifact, always having its type as prefix.
    An artifact identifier is unique within a MLTE context
    (model, version) and for a given artifact type.
    """

    self.timestamp = -1
    """The Unix timestamp of when the artifact was saved to a store."""

    self.creator = None
    """The user that created this artifact."""

    self.level = ArtifactLevel.VERSION
    """The artifact level, be it model or version, defaults to version."""

__str__()

Return a string representation.

Source code in mlte/artifact/artifact.py 
259
260
261
def __str__(self) -> str:
    """Return a string representation."""
    return self.to_model().to_json_string()

build_artifact_header()

Generates the common header model for artifacts.

Source code in mlte/artifact/artifact.py 
245
246
247
248
249
250
251
252
253
254
255
256
257
def build_artifact_header(self) -> ArtifactHeaderModel:
    """Generates the common header model for artifacts."""
    if not self.type:
        raise RuntimeError(
            "Malformed artifact class, type has not been set."
        )
    return ArtifactHeaderModel(
        identifier=self.identifier,
        type=self.type,
        timestamp=self.timestamp,
        creator=self.creator,
        level=self.level,
    )

build_full_id(base=None) classmethod

Builds the full id for this artifact. If base is None, default base is used.

Source code in mlte/artifact/artifact.py 
226
227
228
229
230
231
232
233
234
235
@classmethod
def build_full_id(cls, base: Optional[str] = None) -> str:
    """Builds the full id for this artifact. If base is None, default base is used."""
    if not cls.type:
        raise RuntimeError(
            "Malformed artifact class, type has not been set."
        )
    if not base:
        base = DEFAULT_ID
    return Artifact._build_id(cls.type.value, base)

load(identifier=None) classmethod

Load an artifact from the configured global session.

Parameters:

Name Type Description Default
identifier Optional[str]

The identifier for the artifact. If None, the default id is used. This is equivalent to calling: Artifact.load_with(session().context, session().store)

 None
Source code in mlte/artifact/artifact.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
@classmethod
def load(cls, identifier: Optional[str] = None) -> Artifact:
    """
    Load an artifact from the configured global session.
    :param identifier: The identifier for the artifact. If None,
    the default id is used.

    This is equivalent to calling:
        Artifact.load_with(session().context, session().store)
    """
    return cls.load_with(
        identifier,
        context=get_session().context,
        store=get_session().stores.artifact_store,
    )

load_models(artifact_type, context, store) staticmethod

Loads all artifact models of the given type for the given context and store.

Source code in mlte/artifact/artifact.py 
214
215
216
217
218
219
220
221
222
223
224
@staticmethod
def load_models(
    artifact_type: ArtifactType, context: Context, store: ArtifactStore
) -> list[ArtifactModel]:
    """Loads all artifact models of the given type for the given context and store."""
    with ManagedArtifactSession(store.session()) as artifact_store:
        query_instance = Query(filter=TypeFilter(item_type=artifact_type))
        artifact_models = artifact_store.artifact_mapper.search(
            query_instance, context=(context.model, context.version)
        )
        return artifact_models

load_models_for_session(artifact_type) staticmethod

Loads all artifact models of the given type from the session.

Source code in mlte/artifact/artifact.py 
203
204
205
206
207
208
209
210
211
212
@staticmethod
def load_models_for_session(
    artifact_type: ArtifactType,
) -> list[ArtifactModel]:
    """Loads all artifact models of the given type from the session."""
    return Artifact.load_models(
        artifact_type,
        context=get_session().context,
        store=get_session().stores.artifact_store,
    )

load_with(identifier=None, *, context, store) classmethod

Load an artifact with the given context and store configuration.

Parameters:

Name Type Description Default
identifier Optional[str]

The identifier for the artifact If None, the default id is used.

 None
context Context

The context from which to load the artifact

 required
store ArtifactStore

The store from which to load the artifact

 required
Source code in mlte/artifact/artifact.py 
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
@classmethod
def load_with(
    cls,
    identifier: Optional[str] = None,
    *,
    context: Context,
    store: ArtifactStore,
) -> Artifact:
    """
    Load an artifact with the given context and store configuration.
    :param identifier: The identifier for the artifact If None,
    the default id is used.
    :param context: The context from which to load the artifact
    :param store: The store from which to load the artifact
    """
    identifier = cls.build_full_id(identifier)

    with ManagedArtifactSession(store.session()) as artifact_store:
        artifact = typing.cast(
            Artifact,
            cls.from_model(
                artifact_store.artifact_mapper.read(
                    identifier, (context.model, context.version)
                )
            ),
        )

    artifact.post_load_hook(context, store)
    return artifact

post_load_hook(context, store)

A method that artifact subclasses may override to enforce post-load invariants.

Parameters:

Name Type Description Default
context Context

The context in which to save the artifact

 required
store ArtifactStore

The store in which to save the artifact

 required

Raises:

Type Description
RuntimeError

On broken invariant
Source code in mlte/artifact/artifact.py 
75
76
77
78
79
80
81
82
83
def post_load_hook(self, context: Context, store: ArtifactStore) -> None:
    """
    A method that artifact subclasses may override to enforce post-load invariants.
    :param context: The context in which to save the artifact
    :param store: The store in which to save the artifact
    :raises RuntimeError: On broken invariant
    """
    # Default implementation is a no-op
    pass

pre_save_hook(context, store)

A method that artifact subclasses can override to enforce pre-save invariants.

Parameters:

Name Type Description Default
context Context

The context in which to save the artifact

 required
store ArtifactStore

The store in which to save the artifact

 required

Raises:

Type Description
RuntimeError

On broken invariant
Source code in mlte/artifact/artifact.py 
65
66
67
68
69
70
71
72
73
def pre_save_hook(self, context: Context, store: ArtifactStore) -> None:
    """
    A method that artifact subclasses can override to enforce pre-save invariants.
    :param context: The context in which to save the artifact
    :param store: The store in which to save the artifact
    :raises RuntimeError: On broken invariant
    """
    # Default implementation is a no-op
    pass

save(*, force=False, parents=False, user=None)

Save an artifact with parameters from the configured global session.

This is equivalent to calling: artifact.save_with(get_session().context, get_session().store)

Parameters:

Name Type Description Default
force bool

Indicates that an existing artifact may be overwritten

 False
parents bool

Indicates whether organizational elements for the artifact are created implicitly on write (default: False)

 False
user Optional[str]

The username of the user executing this action.

 None

Returns:

Type Description
ArtifactModel

The ArtifactModel of the saved artifact.
Source code in mlte/artifact/artifact.py 
 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
def save(
    self,
    *,
    force: bool = False,
    parents: bool = False,
    user: Optional[str] = None,
) -> ArtifactModel:
    """
    Save an artifact with parameters from the configured global session.

    This is equivalent to calling:
        artifact.save_with(get_session().context, get_session().store)

    :param force: Indicates that an existing artifact may be overwritten
    :param parents: Indicates whether organizational elements for the
    artifact are created implicitly on write (default: False)
    :param user: The username of the user executing this action.
    :return: The ArtifactModel of the saved artifact.
    """
    credentials = get_session().credentials
    return self.save_with(
        get_session().context,
        get_session().stores.artifact_store,
        force=force,
        parents=parents,
        user=(
            user if user else (credentials.user if credentials else None)
        ),
    )

save_with(context, store, *, force=False, parents=False, user=None)

Save an artifact with the given context and store configuration.

Parameters:

Name Type Description Default
context Context

The context in which to save the artifact

 required
store ArtifactStore

The store in which to save the artifact

 required
force bool

Indicates that an existing artifact may be overwritten

 False
parents bool

Indicates whether organizational elements for the artifact are created implicitly on write (default: False)

 False
user Optional[str]

The username of the user executing this action.

 None

Returns:

Type Description
ArtifactModel

The ArtifactModel of the saved artifact.
Source code in mlte/artifact/artifact.py 
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
def save_with(
    self,
    context: Context,
    store: ArtifactStore,
    *,
    force: bool = False,
    parents: bool = False,
    user: Optional[str] = None,
) -> ArtifactModel:
    """
    Save an artifact with the given context and store configuration.
    :param context: The context in which to save the artifact
    :param store: The store in which to save the artifact
    :param force: Indicates that an existing artifact may be overwritten
    :param parents: Indicates whether organizational elements for the
    artifact are created implicitly on write (default: False)
    :param user: The username of the user executing this action.
    :return: The ArtifactModel of the saved artifact.
    """
    with ManagedArtifactSession(store.session()) as artifact_store:
        # If we are forcing parent creation, ensure they are there before any hooks.
        if parents:
            artifact_store.create_parents(context.model, context.version)

        # Run any artifact-type specific pre save hooks.
        self.pre_save_hook(context, store)

        # Convert to model and save.
        model = self.to_model()

        assert isinstance(
            model, ArtifactModel
        ), "Can't create object from non-ArtifactModel model."

        return artifact_store.artifact_mapper.write_artifact_with_header(
            context.model,
            context.version,
            model,
            force=force,
            user=user,
        )