Skip to content

manifest

Attributes

Classes

Manifest

Bases: KiaraModel

A class to hold the type and configuration for a module instance.

Source code in kiara/models/module/manifest.py 
 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
class Manifest(KiaraModel):

    """A class to hold the type and configuration for a module instance."""

    _kiara_model_id = "instance.manifest"

    class Config:
        extra = Extra.forbid
        validate_all = True

    _manifest_data: Union[Mapping[str, Any], None] = PrivateAttr(default=None)
    _manifest_cid: Union[CID, None] = PrivateAttr(default=None)

    module_type: str = Field(description="The module type.")
    module_config: Mapping[str, Any] = Field(
        default_factory=dict, description="The configuration for the module."
    )
    is_resolved: bool = Field(
        description="Whether the configuration of this module was augmented with the module type defaults etc.",
        default=False,
    )
    # python_class: PythonClass = Field(description="The python class that implements this module.")
    # doc: DocumentationMetadataModel = Field(
    #     description="Documentation for this module instance.", default=None
    # )

    # @validator("module_config")
    # def _validate_module_config(cls, value):
    #
    #     return value

    @property
    def manifest_data(self):
        """The configuration data for this module instance."""
        if self._manifest_data is not None:
            return self._manifest_data

        self._manifest_data = {
            "module_type": self.module_type,
            "module_config": self.module_config,
        }
        return self._manifest_data

    @property
    def manifest_cid(self) -> CID:
        if self._manifest_cid is not None:
            return self._manifest_cid

        _, self._manifest_cid = compute_cid(self.manifest_data)
        return self._manifest_cid

    @property
    def manifest_hash(self) -> str:
        return str(self.manifest_cid)

    def manifest_data_as_json(self):

        return self.json(include={"module_type", "module_config"})

    def _retrieve_data_to_hash(self) -> Any:
        return self.manifest_data

    def create_renderable(self, **config: Any) -> RenderableType:
        """Create a renderable for this module configuration."""
        data = self.dict(exclude_none=True)
        conf = Syntax(
            orjson_dumps(data, option=orjson.OPT_INDENT_2),
            "json",
            background_color="default",
        )
        return conf

    def __repr__(self):

        return f"{self.__class__.__name__}(module_type={self.module_type}, module_config={self.module_config})"

    def __str__(self):

        return self.__repr__()

Attributes

module_type: str = Field(description='The module type.') class-attribute
module_config: Mapping[str, Any] = Field(default_factory=dict, description='The configuration for the module.') class-attribute
is_resolved: bool = Field(description='Whether the configuration of this module was augmented with the module type defaults etc.', default=False) class-attribute
manifest_data property

The configuration data for this module instance.

manifest_cid: CID property
manifest_hash: str property

Classes

Config
Source code in kiara/models/module/manifest.py 
33
34
35
class Config:
    extra = Extra.forbid
    validate_all = True
Attributes
extra = Extra.forbid class-attribute
validate_all = True class-attribute

Functions

manifest_data_as_json()
Source code in kiara/models/module/manifest.py 
82
83
84
def manifest_data_as_json(self):

    return self.json(include={"module_type", "module_config"})
create_renderable(**config: Any) -> RenderableType

Create a renderable for this module configuration.

Source code in kiara/models/module/manifest.py 
89
90
91
92
93
94
95
96
97
def create_renderable(self, **config: Any) -> RenderableType:
    """Create a renderable for this module configuration."""
    data = self.dict(exclude_none=True)
    conf = Syntax(
        orjson_dumps(data, option=orjson.OPT_INDENT_2),
        "json",
        background_color="default",
    )
    return conf

InputsManifest

Bases: Manifest

Source code in kiara/models/module/manifest.py 
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
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
class InputsManifest(Manifest):

    _kiara_model_id = "instance.manifest_with_inputs"

    inputs: Mapping[str, uuid.UUID] = Field(
        description="A map of all the input fields and value references."
    )
    _inputs_cid: Union[CID, None] = PrivateAttr(default=None)
    _jobs_cid: Union[CID, None] = PrivateAttr(default=None)
    _inputs_data_cid: Union[bool, CID, None] = PrivateAttr(default=None)

    @validator("inputs")
    def replace_none_values(cls, value):
        result = {}
        for k, v in value.items():
            if v is None:
                v = NONE_VALUE_ID
            result[k] = v
        return result

    @property
    def job_hash(self) -> str:

        return str(self.job_cid)

    @property
    def job_cid(self) -> CID:

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

        obj: IPLDKind = {"manifest": self.manifest_cid, "inputs": self.inputs_cid}
        _, self._jobs_cid = compute_cid(data=obj)
        return self._jobs_cid

    @property
    def inputs_cid(self) -> CID:
        if self._inputs_cid is not None:
            return self._inputs_cid

        _, cid = compute_cid(data={k: v.bytes for k, v in self.inputs.items()})
        self._inputs_cid = cid
        return self._inputs_cid

    @property
    def inputs_hash(self) -> str:
        return str(self.inputs_cid)

    def calculate_inputs_data_cid(
        self, data_registry: "DataRegistry"
    ) -> Union[CID, None]:

        if self._inputs_data_cid is not None:
            if self._inputs_data_cid is False:
                return None
            return self._inputs_data_cid  # type: ignore

        data_hashes: Dict[str, Any] = {}
        invalid = False

        for k, v in self.inputs.items():
            value = data_registry.get_value(v)
            if value.value_hash == INVALID_HASH_MARKER:
                invalid = True
                break
            data_hashes[k] = CID.decode(value.value_hash)

        if invalid:
            self._inputs_data_cid = False
            return None

        _, cid = compute_cid(data=data_hashes)
        self._inputs_data_cid = cid
        return cid

Attributes

inputs: Mapping[str, uuid.UUID] = Field(description='A map of all the input fields and value references.') class-attribute
job_hash: str property
job_cid: CID property
inputs_cid: CID property
inputs_hash: str property

Functions

replace_none_values(value)
Source code in kiara/models/module/manifest.py 
119
120
121
122
123
124
125
126
@validator("inputs")
def replace_none_values(cls, value):
    result = {}
    for k, v in value.items():
        if v is None:
            v = NONE_VALUE_ID
        result[k] = v
    return result
calculate_inputs_data_cid(data_registry: DataRegistry) -> Union[CID, None]
Source code in kiara/models/module/manifest.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
def calculate_inputs_data_cid(
    self, data_registry: "DataRegistry"
) -> Union[CID, None]:

    if self._inputs_data_cid is not None:
        if self._inputs_data_cid is False:
            return None
        return self._inputs_data_cid  # type: ignore

    data_hashes: Dict[str, Any] = {}
    invalid = False

    for k, v in self.inputs.items():
        value = data_registry.get_value(v)
        if value.value_hash == INVALID_HASH_MARKER:
            invalid = True
            break
        data_hashes[k] = CID.decode(value.value_hash)

    if invalid:
        self._inputs_data_cid = False
        return None

    _, cid = compute_cid(data=data_hashes)
    self._inputs_data_cid = cid
    return cid

Functions