Skip to content

Entity and methods

Artifact

Bases: Entity

A class representing a artifact.

Artifacts are (binary) objects stored in one of the artifact stores of the platform, and available to every process, module and component as files.

Source code in digitalhub_core/entities/artifacts/entity.py 
 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
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
class Artifact(Entity):
    """
    A class representing a artifact.

    Artifacts are (binary) objects stored in one of the artifact
    stores of the platform, and available to every process, module
    and component as files.
    """

    ENTITY_TYPE = EntityTypes.ARTIFACTS.value

    def __init__(
        self,
        project: str,
        name: str,
        uuid: str,
        kind: str,
        metadata: Metadata,
        spec: ArtifactSpec,
        status: ArtifactStatus,
        user: str | None = None,
    ) -> None:
        """
        Constructor.

        Parameters
        ----------
        project : str
            Project name.
        name : str
            Name of the object.
        uuid : str
            Version of the object.
        kind : str
            Kind the object.
        metadata : Metadata
            Metadata of the object.
        spec : ArtifactSpec
            Specification of the object.
        status : ArtifactStatus
            Status of the object.
        user : str
            Owner of the object.
        """
        super().__init__()
        self.project = project
        self.name = name
        self.id = uuid
        self.kind = kind
        self.key = f"store://{project}/{self.ENTITY_TYPE}/{kind}/{name}:{uuid}"
        self.metadata = metadata
        self.spec = spec
        self.status = status
        self.user = user

        # Add attributes to be used in the to_dict method
        self._obj_attr.extend(["project", "name", "id", "key"])

    #############################
    #  Save / Refresh / Export
    #############################

    def save(self, update: bool = False) -> Artifact:
        """
        Save entity into backend.

        Parameters
        ----------
        update : bool
            Flag to indicate update.

        Returns
        -------
        Artifact
            Entity saved.
        """
        obj = self.to_dict()

        if not update:
            new_obj = create_entity_api_ctx(self.project, self.ENTITY_TYPE, obj)
            self._update_attributes(new_obj)
            return self

        self.metadata.updated = obj["metadata"]["updated"] = get_timestamp()
        new_obj = update_entity_api_ctx(self.project, self.ENTITY_TYPE, self.id, obj)
        self._update_attributes(new_obj)
        return self

    def refresh(self) -> Artifact:
        """
        Refresh object from backend.

        Returns
        -------
        Artifact
            Entity refreshed.
        """
        new_obj = read_entity_api_ctx(self.key)
        self._update_attributes(new_obj)
        return self

    def export(self, filename: str | None = None) -> None:
        """
        Export object as a YAML file.

        Parameters
        ----------
        filename : str
            Name of the export YAML file. If not specified, the default value is used.

        Returns
        -------
        None
        """
        obj = self.to_dict()
        if filename is None:
            filename = f"{self.kind}_{self.name}_{self.id}.yml"
        pth = self._context().root / filename
        pth.parent.mkdir(parents=True, exist_ok=True)
        write_yaml(pth, obj)

    #############################
    #  Context
    #############################

    def _context(self) -> Context:
        """
        Get context.

        Returns
        -------
        Context
            Context.
        """
        return get_context(self.project)

    #############################
    #  Artifacts Methods
    #############################

    def as_file(self) -> str:
        """
        Get artifact as file. In the case of a local store, the store returns the current
        path of the artifact. In the case of a remote store, the artifact is downloaded in
        a temporary directory.

        Returns
        -------
        str
            Path of the artifact.
        """
        path = self.spec.path
        store = get_store(path)
        if store.is_local():
            return path
        return store.download(path)

    def download(
        self,
        destination: str | None = None,
        overwrite: bool = False,
    ) -> str:
        """
        Download artifact from remote storage.

        Parameters
        ----------
        destination : str
            Destination path as filename.
        overwrite : bool
            Specify if overwrite an existing file. Default value is False.

        Returns
        -------
        str
            Path of the downloaded artifact.
        """

        # Check if target path is remote
        path = self.spec.path
        store = get_store(path)
        if store.is_local():
            raise RuntimeError("Local files cannot be downloaded. Use as_file().")

        # Check if download destination path is specified and rebuild it if necessary
        if destination is None:
            filename = Path(urlparse(path).path).name
            destination = f"{self.project}/{self.ENTITY_TYPE}/{self.name}/{self.id}/{filename}"

        # Check if destination path is local
        self._check_local(destination)

        # Check if destination path exists for overwrite
        check_overwrite(destination, overwrite)

        # Download dataitem and return path
        return store.download(path, destination)
        return store.download(path, destination)

    def upload(self, source: str | None = None) -> str:
        """
        Upload artifact to remote storage from given local path to
        spec path destination.

        Parameters
        ----------
        source : str
            Source path is the local path of the artifact.

        Returns
        -------
        str
            Path of the uploaded artifact.
        """
        # Check if target path is remote
        path = self.spec.path
        store = get_store(path)
        if store.is_local():
            raise RuntimeError("Only remote paths are supported for upload.")

        # Check if source path is provided and if it is local
        src = source if source is not None else self.spec.src_path
        self._check_local(src)

        # Get store and upload artifact and return remote path
        target = store.upload(src, path)
        file_info = store.get_file_info(target, src)

        if file_info is not None:
            self.refresh()
            self.status.add_file(file_info)
            self.save(update=True)

        return target

    #############################
    #  Private Helpers
    #############################

    @staticmethod
    def _check_local(path: str) -> None:
        """
        Check through URI scheme if given path is local or not.

        Parameters
        ----------
        path : str
            Path of some source.

        Returns
        -------
        None

        Raises
        ------
        EntityError
            If source path is not local.
        """
        if not check_local_path(path):
            raise EntityError("Only local paths are supported for source paths.")

    #############################
    #  Static interface methods
    #############################

    @staticmethod
    def _parse_dict(obj: dict, validate: bool = True) -> dict:
        """
        Get dictionary and parse it to a valid entity dictionary.

        Parameters
        ----------
        entity : str
            Entity type.
        obj : dict
            Dictionary to parse.

        Returns
        -------
        dict
            A dictionary containing the attributes of the entity instance.
        """
        project = obj.get("project")
        name = build_name(obj.get("name"))
        kind = obj.get("kind")
        uuid = build_uuid(obj.get("id"))
        metadata = build_metadata(kind, **obj.get("metadata", {}))
        spec = build_spec(kind, validate=validate, **obj.get("spec", {}))
        status = build_status(kind, **obj.get("status", {}))
        user = obj.get("user")
        return {
            "project": project,
            "name": name,
            "uuid": uuid,
            "kind": kind,
            "metadata": metadata,
            "spec": spec,
            "status": status,
            "user": user,
        }

__init__(project, name, uuid, kind, metadata, spec, status, user=None)

Constructor.

Parameters:

Name Type Description Default
project str

Project name.

 required
name str

Name of the object.

 required
uuid str

Version of the object.

 required
kind str

Kind the object.

 required
metadata Metadata

Metadata of the object.

 required
spec ArtifactSpec

Specification of the object.

 required
status ArtifactStatus

Status of the object.

 required
user str

Owner of the object.

 None
Source code in digitalhub_core/entities/artifacts/entity.py 
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
def __init__(
    self,
    project: str,
    name: str,
    uuid: str,
    kind: str,
    metadata: Metadata,
    spec: ArtifactSpec,
    status: ArtifactStatus,
    user: str | None = None,
) -> None:
    """
    Constructor.

    Parameters
    ----------
    project : str
        Project name.
    name : str
        Name of the object.
    uuid : str
        Version of the object.
    kind : str
        Kind the object.
    metadata : Metadata
        Metadata of the object.
    spec : ArtifactSpec
        Specification of the object.
    status : ArtifactStatus
        Status of the object.
    user : str
        Owner of the object.
    """
    super().__init__()
    self.project = project
    self.name = name
    self.id = uuid
    self.kind = kind
    self.key = f"store://{project}/{self.ENTITY_TYPE}/{kind}/{name}:{uuid}"
    self.metadata = metadata
    self.spec = spec
    self.status = status
    self.user = user

    # Add attributes to be used in the to_dict method
    self._obj_attr.extend(["project", "name", "id", "key"])

as_file()

Get artifact as file. In the case of a local store, the store returns the current path of the artifact. In the case of a remote store, the artifact is downloaded in a temporary directory.

Returns:

Type Description
str

Path of the artifact.
Source code in digitalhub_core/entities/artifacts/entity.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
def as_file(self) -> str:
    """
    Get artifact as file. In the case of a local store, the store returns the current
    path of the artifact. In the case of a remote store, the artifact is downloaded in
    a temporary directory.

    Returns
    -------
    str
        Path of the artifact.
    """
    path = self.spec.path
    store = get_store(path)
    if store.is_local():
        return path
    return store.download(path)

download(destination=None, overwrite=False)

Download artifact from remote storage.

Parameters:

Name Type Description Default
destination str

Destination path as filename.

 None
overwrite bool

Specify if overwrite an existing file. Default value is False.

 False

Returns:

Type Description
str

Path of the downloaded artifact.
Source code in digitalhub_core/entities/artifacts/entity.py 
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
def download(
    self,
    destination: str | None = None,
    overwrite: bool = False,
) -> str:
    """
    Download artifact from remote storage.

    Parameters
    ----------
    destination : str
        Destination path as filename.
    overwrite : bool
        Specify if overwrite an existing file. Default value is False.

    Returns
    -------
    str
        Path of the downloaded artifact.
    """

    # Check if target path is remote
    path = self.spec.path
    store = get_store(path)
    if store.is_local():
        raise RuntimeError("Local files cannot be downloaded. Use as_file().")

    # Check if download destination path is specified and rebuild it if necessary
    if destination is None:
        filename = Path(urlparse(path).path).name
        destination = f"{self.project}/{self.ENTITY_TYPE}/{self.name}/{self.id}/{filename}"

    # Check if destination path is local
    self._check_local(destination)

    # Check if destination path exists for overwrite
    check_overwrite(destination, overwrite)

    # Download dataitem and return path
    return store.download(path, destination)
    return store.download(path, destination)

export(filename=None)

Export object as a YAML file.

Parameters:

Name Type Description Default
filename str

Name of the export YAML file. If not specified, the default value is used.

 None

Returns:

Type Description
None
Source code in digitalhub_core/entities/artifacts/entity.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def export(self, filename: str | None = None) -> None:
    """
    Export object as a YAML file.

    Parameters
    ----------
    filename : str
        Name of the export YAML file. If not specified, the default value is used.

    Returns
    -------
    None
    """
    obj = self.to_dict()
    if filename is None:
        filename = f"{self.kind}_{self.name}_{self.id}.yml"
    pth = self._context().root / filename
    pth.parent.mkdir(parents=True, exist_ok=True)
    write_yaml(pth, obj)

refresh()

Refresh object from backend.

Returns:

Type Description
Artifact

Entity refreshed.
Source code in digitalhub_core/entities/artifacts/entity.py 
117
118
119
120
121
122
123
124
125
126
127
128
def refresh(self) -> Artifact:
    """
    Refresh object from backend.

    Returns
    -------
    Artifact
        Entity refreshed.
    """
    new_obj = read_entity_api_ctx(self.key)
    self._update_attributes(new_obj)
    return self

save(update=False)

Save entity into backend.

Parameters:

Name Type Description Default
update bool

Flag to indicate update.

 False

Returns:

Type Description
Artifact

Entity saved.
Source code in digitalhub_core/entities/artifacts/entity.py 
 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
def save(self, update: bool = False) -> Artifact:
    """
    Save entity into backend.

    Parameters
    ----------
    update : bool
        Flag to indicate update.

    Returns
    -------
    Artifact
        Entity saved.
    """
    obj = self.to_dict()

    if not update:
        new_obj = create_entity_api_ctx(self.project, self.ENTITY_TYPE, obj)
        self._update_attributes(new_obj)
        return self

    self.metadata.updated = obj["metadata"]["updated"] = get_timestamp()
    new_obj = update_entity_api_ctx(self.project, self.ENTITY_TYPE, self.id, obj)
    self._update_attributes(new_obj)
    return self

upload(source=None)

Upload artifact to remote storage from given local path to spec path destination.

Parameters:

Name Type Description Default
source str

Source path is the local path of the artifact.

 None

Returns:

Type Description
str

Path of the uploaded artifact.
Source code in digitalhub_core/entities/artifacts/entity.py 
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
262
def upload(self, source: str | None = None) -> str:
    """
    Upload artifact to remote storage from given local path to
    spec path destination.

    Parameters
    ----------
    source : str
        Source path is the local path of the artifact.

    Returns
    -------
    str
        Path of the uploaded artifact.
    """
    # Check if target path is remote
    path = self.spec.path
    store = get_store(path)
    if store.is_local():
        raise RuntimeError("Only remote paths are supported for upload.")

    # Check if source path is provided and if it is local
    src = source if source is not None else self.spec.src_path
    self._check_local(src)

    # Get store and upload artifact and return remote path
    target = store.upload(src, path)
    file_info = store.get_file_info(target, src)

    if file_info is not None:
        self.refresh()
        self.status.add_file(file_info)
        self.save(update=True)

    return target

artifact_from_dict(obj)

Create artifact from dictionary.

Parameters:

Name Type Description Default
obj dict

Dictionary to create object from.

 required

Returns:

Type Description
Artifact

Artifact object.
Source code in digitalhub_core/entities/artifacts/entity.py 
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
def artifact_from_dict(obj: dict) -> Artifact:
    """
    Create artifact from dictionary.

    Parameters
    ----------
    obj : dict
        Dictionary to create object from.

    Returns
    -------
    Artifact
        Artifact object.
    """
    return Artifact.from_dict(obj)

artifact_from_parameters(project, name, kind, uuid=None, description=None, git_source=None, labels=None, embedded=True, path=None, src_path=None, **kwargs)

Create an instance of the Artifact class with the provided parameters.

Parameters:

Name Type Description Default
project str

Project name.

 required
name str

Object name.

 required
kind str

Kind the object.

 required
uuid str

ID of the object (UUID4).

 None
description str

Description of the object (human readable).

 None
git_source str

Remote git source for object.

 None
labels list[str]

List of labels.

 None
embedded bool

Flag to determine if object must be embedded in project.

 True
path str

Object path on local file system or remote storage. If not provided, it's generated.

 None
src_path str

Local object path.

 None
**kwargs dict

Spec keyword arguments.

 {}

Returns:

Type Description
Artifact

Artifact object.
Source code in digitalhub_core/entities/artifacts/entity.py 
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
def artifact_from_parameters(
    project: str,
    name: str,
    kind: str,
    uuid: str | None = None,
    description: str | None = None,
    git_source: str | None = None,
    labels: list[str] | None = None,
    embedded: bool = True,
    path: str | None = None,
    src_path: str | None = None,
    **kwargs,
) -> Artifact:
    """
    Create an instance of the Artifact class with the provided parameters.

    Parameters
    ----------
    project : str
        Project name.
    name : str
        Object name.
    kind : str
        Kind the object.
    uuid : str
        ID of the object (UUID4).
    description : str
        Description of the object (human readable).
    git_source : str
        Remote git source for object.
    labels : list[str]
        List of labels.
    embedded : bool
        Flag to determine if object must be embedded in project.
    path : str
        Object path on local file system or remote storage.
        If not provided, it's generated.
    src_path : str
        Local object path.
    **kwargs : dict
        Spec keyword arguments.

    Returns
    -------
    Artifact
        Artifact object.
    """
    if path is None:
        raise EntityError("Path must be provided.")
    name = build_name(name)
    uuid = build_uuid(uuid)
    metadata = build_metadata(
        kind,
        project=project,
        name=name,
        version=uuid,
        description=description,
        source=git_source,
        labels=labels,
        embedded=embedded,
    )
    spec = build_spec(
        kind,
        path=path,
        src_path=src_path,
        **kwargs,
    )
    status = build_status(kind)
    return Artifact(
        project=project,
        name=name,
        uuid=uuid,
        kind=kind,
        metadata=metadata,
        spec=spec,
        status=status,
    )