Skip to content

Spec

RunParams

Bases: SpecParams

Run parameters.

Source code in digitalhub_core/entities/run/spec.py 
117
118
119
120
121
122
123
124
125
126
class RunParams(SpecParams):
    """
    Run parameters.
    """

    task: str = None
    """The task string associated with the run."""

    local_execution: bool = False
    """Flag to indicate if the run will be executed locally."""

local_execution: bool = False

Flag to indicate if the run will be executed locally.

task: str = None

The task string associated with the run.

RunSpec

Bases: Spec

Run specification.

Source code in digitalhub_core/entities/run/spec.py 
 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
class RunSpec(Spec):
    """Run specification."""

    def __init__(
        self,
        task: str,
        local_execution: bool = False,
    ) -> None:
        self.task = task
        self.local_execution = local_execution

    def get_inputs(self, as_dict: bool = False) -> dict:
        """
        Get inputs.

        Returns
        -------
        dict
            The inputs.
        """
        inputs = {}
        if not hasattr(self, "inputs") or self.inputs is None:
            return inputs

        for parameter, item in self.inputs.items():
            parameter_type = self._parse_parameter(parameter)

            # Get entity from key
            if parameter_type == "key":
                key = self._collect_key(item)
                entity = self._collect_entity(key)
                if as_dict:
                    entity = entity.to_dict()
                inputs[parameter] = entity

            # Create entity from parameter
            elif parameter_type == "create":
                raise NotImplementedError

        return inputs

    @staticmethod
    def _parse_parameter(parameter: str) -> str:
        """
        Parse parameter.

        Parameters
        ----------
        parameter : str
            Parameter.

        Returns
        -------
        str
            The parsed parameter.
        """
        if len(parameter.split(":")) == 1:
            return "key"
        return "create"

    @staticmethod
    def _collect_key(item: str | dict) -> str:
        """
        Collect key from item.

        Parameters
        ----------
        item : str | dict
            Key or dict representation of the entity.

        Returns
        -------
        str
            The key.
        """
        if isinstance(item, str):
            return item
        return item.get("key")

    @staticmethod
    def _collect_entity(key: str) -> Entity:
        """
        Collect entity from key.

        Parameters
        ----------
        key : str
            Key of the entity.

        Returns
        -------
        Entity
            The entity.
        """
        _, entity_type, _, _, _ = parse_entity_key(key)
        return ENTITY_FUNC[entity_type](key)

get_inputs(as_dict=False)

Get inputs.

Returns:

Type Description
dict

The inputs.
Source code in digitalhub_core/entities/run/spec.py 
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
def get_inputs(self, as_dict: bool = False) -> dict:
    """
    Get inputs.

    Returns
    -------
    dict
        The inputs.
    """
    inputs = {}
    if not hasattr(self, "inputs") or self.inputs is None:
        return inputs

    for parameter, item in self.inputs.items():
        parameter_type = self._parse_parameter(parameter)

        # Get entity from key
        if parameter_type == "key":
            key = self._collect_key(item)
            entity = self._collect_entity(key)
            if as_dict:
                entity = entity.to_dict()
            inputs[parameter] = entity

        # Create entity from parameter
        elif parameter_type == "create":
            raise NotImplementedError

    return inputs

RunParamsData

Bases: RunParams

Run parameters.

Source code in digitalhub_data/entities/run/spec.py 
14
15
16
17
class RunParamsData(RunParams):
    """
    Run parameters.
    """

RunSpecData

Bases: RunSpec

Run specification.

Source code in digitalhub_data/entities/run/spec.py 
10
11
class RunSpecData(RunSpec):
    """Run specification."""

RunParamsMl

Bases: RunParamsData

Run parameters.

Source code in digitalhub_ml/entities/run/spec.py 
14
15
16
17
class RunParamsMl(RunParamsData):
    """
    Run parameters.
    """

RunSpecMl

Bases: RunSpecData

Run specification.

Source code in digitalhub_ml/entities/run/spec.py 
10
11
class RunSpecMl(RunSpecData):
    """Run specification."""