Skip to content

Status

RunStatus

Bases: Status

Status class for run entities.

Source code in digitalhub_core/entities/run/status.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
 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 RunStatus(Status):
    """
    Status class for run entities.
    """

    def __init__(
        self,
        state: str,
        message: str | None = None,
        outputs: list | None = None,
        results: dict | None = None,
        **kwargs,
    ) -> None:
        super().__init__(state, message)
        self.outputs = outputs
        self.results = results

        self._any_setter(**kwargs)

    def get_results(self) -> dict:
        """
        Get results.

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

    def get_outputs(self, as_key: bool = False, as_dict: bool = False) -> dict:
        """
        Get outputs.

        Parameters
        ----------
        as_key : bool
            If True, return outputs as keys.
        as_dict : bool
            If True, return outputs as dictionaries.

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

        for parameter, key in self.outputs.items():
            entity_type = self._get_entity_type(key)
            entity = ENTITY_FUNC[entity_type](key)
            if as_key:
                entity = entity.key
            if as_dict:
                entity = entity.to_dict()
            outputs[parameter] = entity

        return outputs

    @staticmethod
    def _get_entity_type(key: str) -> str:
        """
        Get entity type.

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

        Returns
        -------
        str
            The entity type.
        """
        _, entity_type, _, _, _ = parse_entity_key(key)
        return entity_type

    def get_values(self, values_list: list) -> dict:
        """
        Get values.

        Parameters
        ----------
        values_list : list
            The values list to search in.

        Returns
        -------
        dict
            The values.
        """
        if not hasattr(self, "results") or self.results is None:
            return {}
        return {k: v for k, v in self.get_results().items() if k in values_list}

get_outputs(as_key=False, as_dict=False)

Get outputs.

Parameters:

Name Type Description Default
as_key bool

If True, return outputs as keys.

False
as_dict bool

If True, return outputs as dictionaries.

False

Returns:

Type Description
dict

The outputs.

Source code in digitalhub_core/entities/run/status.py
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
def get_outputs(self, as_key: bool = False, as_dict: bool = False) -> dict:
    """
    Get outputs.

    Parameters
    ----------
    as_key : bool
        If True, return outputs as keys.
    as_dict : bool
        If True, return outputs as dictionaries.

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

    for parameter, key in self.outputs.items():
        entity_type = self._get_entity_type(key)
        entity = ENTITY_FUNC[entity_type](key)
        if as_key:
            entity = entity.key
        if as_dict:
            entity = entity.to_dict()
        outputs[parameter] = entity

    return outputs

get_results()

Get results.

Returns:

Type Description
dict

The results.

Source code in digitalhub_core/entities/run/status.py
36
37
38
39
40
41
42
43
44
45
46
47
def get_results(self) -> dict:
    """
    Get results.

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

get_values(values_list)

Get values.

Parameters:

Name Type Description Default
values_list list

The values list to search in.

required

Returns:

Type Description
dict

The values.

Source code in digitalhub_core/entities/run/status.py
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def get_values(self, values_list: list) -> dict:
    """
    Get values.

    Parameters
    ----------
    values_list : list
        The values list to search in.

    Returns
    -------
    dict
        The values.
    """
    if not hasattr(self, "results") or self.results is None:
        return {}
    return {k: v for k, v in self.get_results().items() if k in values_list}

RunStatusData

Bases: RunStatus

A class representing a run status.

Source code in digitalhub_data/entities/run/status.py
10
11
12
13
class RunStatusData(RunStatus):
    """
    A class representing a run status.
    """

RunStatusMl

Bases: RunStatusData

A class representing a run status.

Source code in digitalhub_ml/entities/run/status.py
10
11
12
13
class RunStatusMl(RunStatusData):
    """
    A class representing a run status.
    """