Skip to content

artifact

mlte/artifact/artifact.py

Artifact protocol implementation.

Artifact

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 
 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
 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
class Artifact(metaclass=abc.ABCMeta):
    """
    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.
    """

    @classmethod
    def __subclasshook__(cls, subclass):
        return meta.has_callables(subclass, "to_model", "from_model")

    def __init__(self, identifier: str, type: ArtifactType) -> None:
        self.identifier = identifier
        """
        The identifier for the artifact.
        An artifact identifier is unique within a MLTE context
        (model, version) and for a given artifact type.
        """

        self.type = type
        """The identifier for the 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."""

    @abc.abstractmethod
    def to_model(self) -> ArtifactModel:
        """Serialize an artifact to its corresponding model."""
        raise NotImplementedError(
            "Artifact.to_model() not implemented for abstract Artifact."
        )

    @classmethod
    @abc.abstractmethod
    def from_model(cls, _: ArtifactModel) -> Artifact:
        """Deserialize an artifact from its corresponding model."""
        raise NotImplementedError(
            "Artifact.from_model() not implemented for abstract Artifact."
        )

    def __json__(self):
        """Hack method to make Artifacts serializable to JSON if importing json-fix before json.dumps."""
        return self.to_model().to_json()

    def _equal(a: Artifact, b: Artifact) -> bool:
        """
        Compare Artifact instances for equality.

        :param a: Input instance
        :param b: Input instance
        :return: `True` if `a` and `b` are equal, `False` otherwise
        """
        return a.to_model() == b.to_model()

    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) -> None:
        """
        Save an artifact with parameters from the configured global session.

        This is equivalent to calling:
            artifact.save_with(session().context, 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)
        """
        self.save_with(
            session().context,
            session().artifact_store,
            force=force,
            parents=parents,
        )

    def save_with(
        self,
        context: Context,
        store: ArtifactStore,
        *,
        force: bool = False,
        parents: bool = False,
    ) -> None:
        """
        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)
        """
        self.pre_save_hook(context, store)

        artifact_model = self.to_model()
        with ManagedArtifactSession(store.session()) as handle:
            handle.write_artifact_with_header(
                context.model,
                context.version,
                artifact_model,
                force=force,
                parents=parents,
            )

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

        This is equivalent to calling:
            Artifact.load_with(session().context, session().store)
        """
        return cls.load_with(
            identifier,
            context=session().context,
            store=session().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
        :param context: The context from which to load the artifact
        :param store: The store from which to load the artifact
        """
        if identifier is None:
            identifier = cls.get_default_id()

        with ManagedArtifactSession(store.session()) as handle:
            artifact = cls.from_model(
                handle.read_artifact(
                    context.model,
                    context.version,
                    identifier,
                )
            )

        artifact.post_load_hook(context, store)
        return artifact

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

    @staticmethod
    def load_all_models_with(
        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 handle:
            query_instance = Query(filter=TypeFilter(item_type=artifact_type))
            artifact_models = handle.search_artifacts(
                context.model,
                context.version,
                query_instance,
            )
            return artifact_models

    @staticmethod
    def get_default_id() -> str:
        """To be overriden by derived classes."""
        return "default"

    def build_artifact_header(self) -> ArtifactHeaderModel:
        """Generates the common header model for artifacts."""
        return ArtifactHeaderModel(
            identifier=self.identifier,
            type=self.type,
            timestamp=self.timestamp,
            creator=self.creator,
        )

creator = None instance-attribute

The user that created this artifact.

identifier = identifier instance-attribute

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

timestamp = -1 instance-attribute

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

type = type instance-attribute

The identifier for the artifact type

__eq__(other)

Test instance for equality.

Source code in mlte/artifact/artifact.py 
82
83
84
85
86
def __eq__(self, other: object) -> bool:
    """Test instance for equality."""
    if not isinstance(other, Artifact):
        return False
    return self._equal(other)

__json__()

Hack method to make Artifacts serializable to JSON if importing json-fix before json.dumps.

Source code in mlte/artifact/artifact.py 
68
69
70
def __json__(self):
    """Hack method to make Artifacts serializable to JSON if importing json-fix before json.dumps."""
    return self.to_model().to_json()

build_artifact_header()

Generates the common header model for artifacts.

Source code in mlte/artifact/artifact.py 
226
227
228
229
230
231
232
233
def build_artifact_header(self) -> ArtifactHeaderModel:
    """Generates the common header model for artifacts."""
    return ArtifactHeaderModel(
        identifier=self.identifier,
        type=self.type,
        timestamp=self.timestamp,
        creator=self.creator,
    )

from_model(_) abstractmethod classmethod

Deserialize an artifact from its corresponding model.

Source code in mlte/artifact/artifact.py 
60
61
62
63
64
65
66
@classmethod
@abc.abstractmethod
def from_model(cls, _: ArtifactModel) -> Artifact:
    """Deserialize an artifact from its corresponding model."""
    raise NotImplementedError(
        "Artifact.from_model() not implemented for abstract Artifact."
    )

get_default_id() staticmethod

To be overriden by derived classes.

Source code in mlte/artifact/artifact.py 
221
222
223
224
@staticmethod
def get_default_id() -> str:
    """To be overriden by derived classes."""
    return "default"

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 This is equivalent to calling: Artifact.load_with(session().context, session().store)

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

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

load_all_models(artifact_type) staticmethod

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

Source code in mlte/artifact/artifact.py 
198
199
200
201
202
203
204
205
@staticmethod
def load_all_models(artifact_type: ArtifactType) -> list[ArtifactModel]:
    """Loads all artifact models of the given type from the session."""
    return Artifact.load_all_models_with(
        artifact_type,
        context=session().context,
        store=session().artifact_store,
    )

load_all_models_with(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 
207
208
209
210
211
212
213
214
215
216
217
218
219
@staticmethod
def load_all_models_with(
    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 handle:
        query_instance = Query(filter=TypeFilter(item_type=artifact_type))
        artifact_models = handle.search_artifacts(
            context.model,
            context.version,
            query_instance,
        )
        return artifact_models

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

 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 
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
@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
    :param context: The context from which to load the artifact
    :param store: The store from which to load the artifact
    """
    if identifier is None:
        identifier = cls.get_default_id()

    with ManagedArtifactSession(store.session()) as handle:
        artifact = cls.from_model(
            handle.read_artifact(
                context.model,
                context.version,
                identifier,
            )
        )

    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 
 98
 99
100
101
102
103
104
105
106
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 
88
89
90
91
92
93
94
95
96
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)

Save an artifact with parameters from the configured global session.

This is equivalent to calling: artifact.save_with(session().context, 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
Source code in mlte/artifact/artifact.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def save(self, *, force: bool = False, parents: bool = False) -> None:
    """
    Save an artifact with parameters from the configured global session.

    This is equivalent to calling:
        artifact.save_with(session().context, 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)
    """
    self.save_with(
        session().context,
        session().artifact_store,
        force=force,
        parents=parents,
    )

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

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
Source code in mlte/artifact/artifact.py 
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 save_with(
    self,
    context: Context,
    store: ArtifactStore,
    *,
    force: bool = False,
    parents: bool = False,
) -> None:
    """
    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)
    """
    self.pre_save_hook(context, store)

    artifact_model = self.to_model()
    with ManagedArtifactSession(store.session()) as handle:
        handle.write_artifact_with_header(
            context.model,
            context.version,
            artifact_model,
            force=force,
            parents=parents,
        )

to_model() abstractmethod

Serialize an artifact to its corresponding model.

Source code in mlte/artifact/artifact.py 
53
54
55
56
57
58
@abc.abstractmethod
def to_model(self) -> ArtifactModel:
    """Serialize an artifact to its corresponding model."""
    raise NotImplementedError(
        "Artifact.to_model() not implemented for abstract Artifact."
    )