Skip to content

runtime_environment

Attributes

logger = structlog.get_logger() module-attribute

Classes

RuntimeEnvironment

Bases: KiaraModel

Source code in kiara/models/runtime_environment/__init__.py
 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
class RuntimeEnvironment(KiaraModel):
    class Config:
        underscore_attrs_are_private = False
        allow_mutation = False

    @classmethod
    def get_environment_type_name(cls) -> str:

        env_type = cls.__fields__["environment_type"]
        args = get_args(env_type.type_)
        assert len(args) == 1

        return args[0]

    @classmethod
    def create_environment_model(cls):

        try:
            type_name = cls.get_environment_type_name()
            data = cls.retrieve_environment_data()
            assert (
                "environment_type" not in data.keys()
                or data["environment_keys"] == type_name
            )
            data["environment_type"] = type_name

        except Exception as e:
            raise Exception(f"Can't create environment model for '{cls.__name__}': {e}")
        return cls(**data)

    def get_category_alias(self) -> str:
        return f"{ENVIRONMENT_TYPE_CATEGORY_ID}.{self.environment_type}"  # type: ignore

    @classmethod
    @abstractmethod
    def retrieve_environment_data(cls) -> Dict[str, Any]:
        pass

    _env_hashes: Union[Mapping[str, str], None] = PrivateAttr(default=None)

    def _create_renderable_for_field(
        self, field_name: str, for_summary: bool = False
    ) -> Union[RenderableType, None]:

        return extract_renderable(getattr(self, field_name))

    def _retrieve_id(self) -> str:
        return self.__class__.get_environment_type_name()

    @property
    def env_hashes(self) -> Mapping[str, str]:

        if self._env_hashes is not None:
            return self._env_hashes

        result = {}
        for k, v in self._retrieve_sub_profile_env_data().items():
            _, cid = compute_cid(data=v)
            result[k] = str(cid)

        self._env_hashes = result
        return self._env_hashes

    def _retrieve_sub_profile_env_data(self) -> Mapping[str, Any]:
        """Return a dictionary with data that identifies one hash profile per key.

        In most cases, this will only return a single-key dictionary containing all the data in that environment. But
        in some cases one might want to have several sub-hash profiles, for example a list of Python packages with and
        without version information, to have more fine-grained control about when to consider two environments functionally
        equal.
        """

        return {DEFAULT_ENV_HASH_KEY: self._retrieve_data_to_hash()}

    def create_renderable(self, **config: Any) -> RenderableType:

        summary = config.get("summary", False)

        table = Table(show_header=False, box=box.SIMPLE)
        table.add_column("field")
        table.add_column("summary")

        hashes_str = orjson_dumps(self.env_hashes, option=orjson.OPT_INDENT_2)
        table.add_row(
            "environment hashes", Syntax(hashes_str, "json", background_color="default")
        )

        for field_name, field in self.__fields__.items():
            summary_item = self._create_renderable_for_field(
                field_name, for_summary=summary
            )
            if summary_item is not None:
                table.add_row(field_name, summary_item)

        return table

Attributes

env_hashes: Mapping[str, str] property

Classes

Config
Source code in kiara/models/runtime_environment/__init__.py
26
27
28
class Config:
    underscore_attrs_are_private = False
    allow_mutation = False
Attributes
underscore_attrs_are_private = False class-attribute
allow_mutation = False class-attribute

Functions

get_environment_type_name() -> str classmethod
Source code in kiara/models/runtime_environment/__init__.py
30
31
32
33
34
35
36
37
@classmethod
def get_environment_type_name(cls) -> str:

    env_type = cls.__fields__["environment_type"]
    args = get_args(env_type.type_)
    assert len(args) == 1

    return args[0]
create_environment_model() classmethod
Source code in kiara/models/runtime_environment/__init__.py
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
@classmethod
def create_environment_model(cls):

    try:
        type_name = cls.get_environment_type_name()
        data = cls.retrieve_environment_data()
        assert (
            "environment_type" not in data.keys()
            or data["environment_keys"] == type_name
        )
        data["environment_type"] = type_name

    except Exception as e:
        raise Exception(f"Can't create environment model for '{cls.__name__}': {e}")
    return cls(**data)
get_category_alias() -> str
Source code in kiara/models/runtime_environment/__init__.py
55
56
def get_category_alias(self) -> str:
    return f"{ENVIRONMENT_TYPE_CATEGORY_ID}.{self.environment_type}"  # type: ignore
retrieve_environment_data() -> Dict[str, Any] classmethod abstractmethod
Source code in kiara/models/runtime_environment/__init__.py
58
59
60
61
@classmethod
@abstractmethod
def retrieve_environment_data(cls) -> Dict[str, Any]:
    pass
create_renderable(**config: Any) -> RenderableType
Source code in kiara/models/runtime_environment/__init__.py
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def create_renderable(self, **config: Any) -> RenderableType:

    summary = config.get("summary", False)

    table = Table(show_header=False, box=box.SIMPLE)
    table.add_column("field")
    table.add_column("summary")

    hashes_str = orjson_dumps(self.env_hashes, option=orjson.OPT_INDENT_2)
    table.add_row(
        "environment hashes", Syntax(hashes_str, "json", background_color="default")
    )

    for field_name, field in self.__fields__.items():
        summary_item = self._create_renderable_for_field(
            field_name, for_summary=summary
        )
        if summary_item is not None:
            table.add_row(field_name, summary_item)

    return table

Functions