Skip to content

Entity and methods

Task

Bases: UnversionedEntity

A class representing a task.

Source code in digitalhub_core/entities/task/entity.py 
 16
 17
 18
 19
 20
 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
class Task(UnversionedEntity):
    """
    A class representing a task.
    """

    ENTITY_TYPE = EntityTypes.TASK.value

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

    ##############################
    #  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)

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/task/entity.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def delete_run(self, entity_key: str) -> None:
    """
    Delete run.

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

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

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/task/entity.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
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/task/entity.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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)

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/task/entity.py 
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
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,
    )