Skip to content

Entity and methods

Function

Bases: ExecutableEntity

A class representing a function.

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

    ENTITY_TYPE = EntityTypes.FUNCTION.value

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

        self.spec: FunctionSpec
        self.status: FunctionStatus

    ##############################
    #  Function Methods
    ##############################

    def run(
        self,
        action: str,
        local_execution: bool = False,
        wait: bool = False,
        log_info: bool = True,
        **kwargs,
    ) -> Run:
        """
        Run function. This method creates a new run and executes it.

        Parameters
        ----------
        action : str
            Action to execute.
        local_execution : bool
            Flag to determine if object has local execution.
        wait : bool
            Flag to wait for execution.
        log_info : bool
            Flag to log information while waiting.
        **kwargs : dict
            Keyword arguments passed to Task and Run builders.

        Returns
        -------
        Run
            Run instance.
        """
        # 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.new_task(task_kind, **kwargs)

        # Run function from task
        run = task.run(run_kind, local_execution, **kwargs)

        # If execution is done by DHCore backend, return the object
        if not local_execution:
            if self._context().local:
                raise BackendError("Cannot run remote function with local backend.")
            if wait:
                return run.wait(log_info=log_info)
            return run

        # If local execution, build and launch run.
        # Detach the run from the main thread
        run.build()
        with ThreadPoolExecutor(max_workers=1) as executor:
            result = executor.submit(run.run)
            r = result.result()
        return r

    def _get_function_string(self) -> str:
        """
        Get function string.

        Returns
        -------
        str
            Function string.
        """
        return self._get_executable_string()

run(action, local_execution=False, wait=False, log_info=True, **kwargs)

Run function. This method creates a new run and executes it.

Parameters:

Name Type Description Default
action str

Action to execute.

 required
local_execution bool

Flag to determine if object has local execution.

 False
wait bool

Flag to wait for execution.

 False
log_info bool

Flag to log information while waiting.

 True
**kwargs dict

Keyword arguments passed to Task and Run builders.

 {}

Returns:

Type Description
Run

Run instance.
Source code in digitalhub_core/entities/function/entity.py 
 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
def run(
    self,
    action: str,
    local_execution: bool = False,
    wait: bool = False,
    log_info: bool = True,
    **kwargs,
) -> Run:
    """
    Run function. This method creates a new run and executes it.

    Parameters
    ----------
    action : str
        Action to execute.
    local_execution : bool
        Flag to determine if object has local execution.
    wait : bool
        Flag to wait for execution.
    log_info : bool
        Flag to log information while waiting.
    **kwargs : dict
        Keyword arguments passed to Task and Run builders.

    Returns
    -------
    Run
        Run instance.
    """
    # 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.new_task(task_kind, **kwargs)

    # Run function from task
    run = task.run(run_kind, local_execution, **kwargs)

    # If execution is done by DHCore backend, return the object
    if not local_execution:
        if self._context().local:
            raise BackendError("Cannot run remote function with local backend.")
        if wait:
            return run.wait(log_info=log_info)
        return run

    # If local execution, build and launch run.
    # Detach the run from the main thread
    run.build()
    with ThreadPoolExecutor(max_workers=1) as executor:
        result = executor.submit(run.run)
        r = result.result()
    return r