Skip to content

Entity and methods

Task

Bases: Entity

A class representing a task.

Source code in digitalhub_core/entities/tasks/entity.py 
 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
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
class Task(Entity):
    """
    A class representing a task.
    """

    ENTITY_TYPE = EntityTypes.TASKS.value

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

        Parameters
        ----------
        project : str
            Project name.
        uuid : str
            UUID.
        kind : str
            Kind the object.
        metadata : Metadata
            Metadata of the object.
        spec : TaskSpec
            Specification of the object.
        status : TaskStatus
            Status of the object.
        user : str
            Owner of the object.
        """
        super().__init__()
        self.project = project
        self.id = uuid
        self.kind = kind
        self.key = f"store://{project}/{self.ENTITY_TYPE}/{kind}/{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", "id", "key"])

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

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

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

        Returns
        -------
        Task
            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) -> Task:
        """
        Refresh object from backend.

        Returns
        -------
        Task
            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)

    #############################
    #  Task methods
    #############################

    def run(
        self,
        run_kind: str,
        local_execution: bool = False,
        **kwargs,
    ) -> Run:
        """
        Run task.

        Parameters
        ----------
        run_kind : str
            Kind the object.
        local_execution : bool
            Flag to indicate if the run will be executed locally.
        **kwargs : dict
            Keyword arguments.

        Returns
        -------
        Run
            Run object.
        """
        return self.new_run(
            project=self.project,
            task=self._get_task_string(),
            kind=run_kind,
            local_execution=local_execution,
            **kwargs,
        )

    def _get_task_string(self) -> str:
        """
        Get task string.

        Returns
        -------
        str
            Task string.
        """
        splitted = self.spec.function.split("://")
        return f"{self.kind}://{splitted[1]}"

    #############################
    # CRUD Methods for Run
    #############################

    def new_run(self, **kwargs) -> Run:
        """
        Create a new run.

        Parameters
        ----------
        **kwargs : dict
            Keyword arguments.

        Returns
        -------
        Run
            Run object.
        """
        if kwargs["local_execution"]:
            return run_from_parameters(**kwargs)
        return new_run(**kwargs)

    def get_run(self, entity_key: str) -> Run:
        """
        Get run.

        Parameters
        ----------
        entity_key : str
            Entity key.

        Returns
        -------
        Run
            Run object.
        """
        return get_run(entity_key)

    def delete_run(self, entity_key: str) -> None:
        """
        Delete run.

        Parameters
        ----------
        entity_key : str
            Entity key.

        Returns
        -------
        None
        """
        delete_run(entity_key)

    #############################
    #  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")
        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,
            "uuid": uuid,
            "kind": kind,
            "metadata": metadata,
            "spec": spec,
            "status": status,
            "user": user,
        }

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

Constructor.

Parameters:

Name Type Description Default
project str

Project name.

 required
uuid str

UUID.

 required
kind str

Kind the object.

 required
metadata Metadata

Metadata of the object.

 required
spec TaskSpec

Specification of the object.

 required
status TaskStatus

Status of the object.

 required
user str

Owner of the object.

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

    Parameters
    ----------
    project : str
        Project name.
    uuid : str
        UUID.
    kind : str
        Kind the object.
    metadata : Metadata
        Metadata of the object.
    spec : TaskSpec
        Specification of the object.
    status : TaskStatus
        Status of the object.
    user : str
        Owner of the object.
    """
    super().__init__()
    self.project = project
    self.id = uuid
    self.kind = kind
    self.key = f"store://{project}/{self.ENTITY_TYPE}/{kind}/{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", "id", "key"])

delete_run(entity_key)

Delete run.

Parameters:

Name Type Description Default
entity_key str

Entity key.

 required

Returns:

Type Description
None
Source code in digitalhub_core/entities/tasks/entity.py 
238
239
240
241
242
243
244
245
246
247
248
249
250
251
def delete_run(self, entity_key: str) -> None:
    """
    Delete run.

    Parameters
    ----------
    entity_key : str
        Entity key.

    Returns
    -------
    None
    """
    delete_run(entity_key)

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/tasks/entity.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
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)

get_run(entity_key)

Get run.

Parameters:

Name Type Description Default
entity_key str

Entity key.

 required

Returns:

Type Description
Run

Run object.
Source code in digitalhub_core/entities/tasks/entity.py 
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
def get_run(self, entity_key: str) -> Run:
    """
    Get run.

    Parameters
    ----------
    entity_key : str
        Entity key.

    Returns
    -------
    Run
        Run object.
    """
    return get_run(entity_key)

new_run(**kwargs)

Create a new run.

Parameters:

Name Type Description Default
**kwargs dict

Keyword arguments.

 {}

Returns:

Type Description
Run

Run object.
Source code in digitalhub_core/entities/tasks/entity.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
def new_run(self, **kwargs) -> Run:
    """
    Create a new run.

    Parameters
    ----------
    **kwargs : dict
        Keyword arguments.

    Returns
    -------
    Run
        Run object.
    """
    if kwargs["local_execution"]:
        return run_from_parameters(**kwargs)
    return new_run(**kwargs)

refresh()

Refresh object from backend.

Returns:

Type Description
Task

Entity refreshed.
Source code in digitalhub_core/entities/tasks/entity.py 
105
106
107
108
109
110
111
112
113
114
115
116
def refresh(self) -> Task:
    """
    Refresh object from backend.

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

run(run_kind, local_execution=False, **kwargs)

Run task.

Parameters:

Name Type Description Default
run_kind str

Kind the object.

 required
local_execution bool

Flag to indicate if the run will be executed locally.

 False
**kwargs dict

Keyword arguments.

 {}

Returns:

Type Description
Run

Run object.
Source code in digitalhub_core/entities/tasks/entity.py 
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
def run(
    self,
    run_kind: str,
    local_execution: bool = False,
    **kwargs,
) -> Run:
    """
    Run task.

    Parameters
    ----------
    run_kind : str
        Kind the object.
    local_execution : bool
        Flag to indicate if the run will be executed locally.
    **kwargs : dict
        Keyword arguments.

    Returns
    -------
    Run
        Run object.
    """
    return self.new_run(
        project=self.project,
        task=self._get_task_string(),
        kind=run_kind,
        local_execution=local_execution,
        **kwargs,
    )

save(update=False)

Save entity into backend.

Parameters:

Name Type Description Default
update bool

Flag to indicate update.

 False

Returns:

Type Description
Task

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

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

    Returns
    -------
    Task
        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

task_from_dict(obj)

Create Task object from dictionary.

Parameters:

Name Type Description Default
obj dict

Dictionary representation of Task.

 required

Returns:

Type Description
Task

Task object.
Source code in digitalhub_core/entities/tasks/entity.py 
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
def task_from_dict(obj: dict) -> Task:
    """
    Create Task object from dictionary.

    Parameters
    ----------
    obj : dict
        Dictionary representation of Task.

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

task_from_parameters(project, kind, uuid=None, git_source=None, labels=None, **kwargs)

Create a new object instance.

Parameters:

Name Type Description Default
project str

Project name.

 required
kind str

Kind the object.

 required
uuid str

ID of the object (UUID4).

 None
git_source str

Remote git source for object.

 None
labels list[str]

List of labels.

 None
**kwargs dict

Spec keyword arguments.

 {}

Returns:

Type Description
Task

Object instance.
Source code in digitalhub_core/entities/tasks/entity.py 
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
329
330
331
332
333
334
335
336
337
338
339
340
341
def task_from_parameters(
    project: str,
    kind: str,
    uuid: str | None = None,
    git_source: str | None = None,
    labels: list[str] | None = None,
    **kwargs,
) -> Task:
    """
    Create a new object instance.

    Parameters
    ----------
    project : str
        Project name.
    kind : str
        Kind the object.
    uuid : str
        ID of the object (UUID4).
    git_source : str
        Remote git source for object.
    labels : list[str]
        List of labels.
    **kwargs : dict
        Spec keyword arguments.

    Returns
    -------
    Task
        Object instance.
    """
    uuid = build_uuid(uuid)
    metadata = build_metadata(
        kind=kind,
        project=project,
        name=uuid,
        source=git_source,
        labels=labels,
    )
    spec = build_spec(kind, **kwargs)
    status = build_status(kind)
    return Task(
        project=project,
        uuid=uuid,
        kind=kind,
        metadata=metadata,
        spec=spec,
        status=status,
    )