Skip to content

Spec

RunParams

Bases: SpecParams, K8s

Run parameters.

Source code in digitalhub/entities/run/spec.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
class RunParams(SpecParams, K8s):
    """
    Run parameters.
    """

    function: str = None
    """The function associated with the run."""

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

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

function: str = None

The function associated with the run.

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/entities/run/spec.py 
 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
136
137
138
class RunSpec(Spec):
    """Run specification."""

    def __init__(
        self,
        task: str,
        local_execution: bool = False,
        function: str | None = None,
        node_selector: dict | None = None,
        volumes: list | None = None,
        resources: dict | None = None,
        affinity: dict | None = None,
        tolerations: list | None = None,
        envs: list | None = None,
        secrets: list | None = None,
        profile: str | None = None,
        **kwargs,
    ) -> None:
        self.task = task
        self.local_execution = local_execution
        self.function = function
        self.node_selector = node_selector
        self.volumes = volumes
        self.resources = resources
        self.affinity = affinity
        self.tolerations = tolerations
        self.envs = envs
        self.secrets = secrets
        self.profile = profile

    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/entities/run/spec.py 
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
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

RunParams

Bases: SpecParams, K8s

Run parameters.

Source code in digitalhub/entities/run/spec.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
class RunParams(SpecParams, K8s):
    """
    Run parameters.
    """

    function: str = None
    """The function associated with the run."""

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

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

function: str = None

The function associated with the run.

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/entities/run/spec.py 
 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
136
137
138
class RunSpec(Spec):
    """Run specification."""

    def __init__(
        self,
        task: str,
        local_execution: bool = False,
        function: str | None = None,
        node_selector: dict | None = None,
        volumes: list | None = None,
        resources: dict | None = None,
        affinity: dict | None = None,
        tolerations: list | None = None,
        envs: list | None = None,
        secrets: list | None = None,
        profile: str | None = None,
        **kwargs,
    ) -> None:
        self.task = task
        self.local_execution = local_execution
        self.function = function
        self.node_selector = node_selector
        self.volumes = volumes
        self.resources = resources
        self.affinity = affinity
        self.tolerations = tolerations
        self.envs = envs
        self.secrets = secrets
        self.profile = profile

    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/entities/run/spec.py 
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
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

RunParams

Bases: SpecParams, K8s

Run parameters.

Source code in digitalhub/entities/run/spec.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
class RunParams(SpecParams, K8s):
    """
    Run parameters.
    """

    function: str = None
    """The function associated with the run."""

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

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

function: str = None

The function associated with the run.

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/entities/run/spec.py 
 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
136
137
138
class RunSpec(Spec):
    """Run specification."""

    def __init__(
        self,
        task: str,
        local_execution: bool = False,
        function: str | None = None,
        node_selector: dict | None = None,
        volumes: list | None = None,
        resources: dict | None = None,
        affinity: dict | None = None,
        tolerations: list | None = None,
        envs: list | None = None,
        secrets: list | None = None,
        profile: str | None = None,
        **kwargs,
    ) -> None:
        self.task = task
        self.local_execution = local_execution
        self.function = function
        self.node_selector = node_selector
        self.volumes = volumes
        self.resources = resources
        self.affinity = affinity
        self.tolerations = tolerations
        self.envs = envs
        self.secrets = secrets
        self.profile = profile

    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/entities/run/spec.py 
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
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