Skip to content

Entity and methods

Workflow

Bases: ExecutableEntity

A class representing a workflow.

Source code in digitalhub/entities/workflow/entity.py 
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
class Workflow(ExecutableEntity):
    """
    A class representing a workflow.
    """

    ENTITY_TYPE = EntityTypes.WORKFLOW.value

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

        self.spec: WorkflowSpec
        self.status: WorkflowStatus

    ##############################
    #  Workflow Methods
    ##############################

    def run(self, action: str | None = None, **kwargs) -> Run:
        """
        Run workflow.

        Parameters
        ----------
        action : str
            Action to execute.
        **kwargs : dict
            Keyword arguments passed to Run builder.

        Returns
        -------
        Run
            Run instance.
        """
        if action is None:
            action = "pipeline"

        # Get kind registry
        kind_reg = get_kind_registry(self.kind)

        # Get task and run kind
        task_kind = kind_reg.get_task_kind_from_action(action=action)
        run_kind = kind_reg.get_run_kind()

        # Create or update new task
        task = self._get_or_create_task(task_kind)

        # Raise error if execution is not done by DHCore backend
        if self._context().local:
            raise BackendError("Cannot run workflow with local backend.")

        return task.run(run_kind, local_execution=False, **kwargs)

run(action=None, **kwargs)

Run workflow.

Parameters:

Name Type Description Default
action str

Action to execute.

 None
**kwargs dict

Keyword arguments passed to Run builder.

 {}

Returns:

Type Description
Run

Run instance.
Source code in digitalhub/entities/workflow/entity.py 
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
def run(self, action: str | None = None, **kwargs) -> Run:
    """
    Run workflow.

    Parameters
    ----------
    action : str
        Action to execute.
    **kwargs : dict
        Keyword arguments passed to Run builder.

    Returns
    -------
    Run
        Run instance.
    """
    if action is None:
        action = "pipeline"

    # Get kind registry
    kind_reg = get_kind_registry(self.kind)

    # Get task and run kind
    task_kind = kind_reg.get_task_kind_from_action(action=action)
    run_kind = kind_reg.get_run_kind()

    # Create or update new task
    task = self._get_or_create_task(task_kind)

    # Raise error if execution is not done by DHCore backend
    if self._context().local:
        raise BackendError("Cannot run workflow with local backend.")

    return task.run(run_kind, local_execution=False, **kwargs)