Skip to content

mock

Attributes

Classes

MockOutput

Bases: BaseModel

Source code in src/kiara/modules/included_core_modules/mock.py 
14
15
16
class MockOutput(BaseModel):
    field_schema: Dict[str, Any] = Field(description="The schema of the output.")
    data: Any = Field(description="The data of the output.", default="mock result data")

Attributes

field_schema: Dict[str, Any] = Field(description='The schema of the output.') class-attribute instance-attribute
data: Any = Field(description='The data of the output.', default='mock result data') class-attribute instance-attribute

MockModuleConfig

Bases: KiaraModuleConfig

Source code in src/kiara/modules/included_core_modules/mock.py 
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
class MockModuleConfig(KiaraModuleConfig):
    _kiara_model_id: ClassVar = "instance.module_config.mock"

    @classmethod
    def create_pipeline_config(
        cls, title: str, description: str, author: str, *steps: "MockModuleConfig"
    ) -> PipelineConfig:
        data: Dict[str, Any] = {
            "pipeline_name": slugify(title),
            "doc": description,
            "context": {"authors": [author]},
            "steps": [],
        }
        for step in steps:
            step_data = {
                "step_id": slugify(step.title),
                "module_type": "dummy",
                "module_config": {
                    "title": step.title,
                    "inputs_schema": step.inputs_schema,
                    "outputs": step.outputs,
                    "desc": step.desc,
                },
            }
            data["steps"].append(step_data)

        pipeline_config = PipelineConfig.from_config(data)
        return pipeline_config

    inputs_schema: Dict[str, Dict[str, Any]] = Field(
        description="The input fields and their types.",
    )

    outputs: Dict[str, MockOutput] = Field(
        description="The outputs fields of the operation, along with their types and mock data.",
        default_factory=default_mock_output,
    )

    title: str = Field(
        description="The title of this operation.", default="mock_operation"
    )
    desc: str = Field(
        description="A description of what this step does.",
        default=DEFAULT_NO_DESC_VALUE,
    )

Attributes

inputs_schema: Dict[str, Dict[str, Any]] = Field(description='The input fields and their types.') class-attribute instance-attribute
outputs: Dict[str, MockOutput] = Field(description='The outputs fields of the operation, along with their types and mock data.', default_factory=default_mock_output) class-attribute instance-attribute
title: str = Field(description='The title of this operation.', default='mock_operation') class-attribute instance-attribute
desc: str = Field(description='A description of what this step does.', default=DEFAULT_NO_DESC_VALUE) class-attribute instance-attribute

Functions

create_pipeline_config(title: str, description: str, author: str, *steps: MockModuleConfig) -> PipelineConfig classmethod
Source code in src/kiara/modules/included_core_modules/mock.py 
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
@classmethod
def create_pipeline_config(
    cls, title: str, description: str, author: str, *steps: "MockModuleConfig"
) -> PipelineConfig:
    data: Dict[str, Any] = {
        "pipeline_name": slugify(title),
        "doc": description,
        "context": {"authors": [author]},
        "steps": [],
    }
    for step in steps:
        step_data = {
            "step_id": slugify(step.title),
            "module_type": "dummy",
            "module_config": {
                "title": step.title,
                "inputs_schema": step.inputs_schema,
                "outputs": step.outputs,
                "desc": step.desc,
            },
        }
        data["steps"].append(step_data)

    pipeline_config = PipelineConfig.from_config(data)
    return pipeline_config

MockKiaraModule

Bases: KiaraModule

Source code in src/kiara/modules/included_core_modules/mock.py 
 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
class MockKiaraModule(KiaraModule):
    _module_type_name = "mock"
    _config_cls = MockModuleConfig

    def create_inputs_schema(
        self,
    ) -> ValueMapSchema:
        result = {}
        v: Mapping[str, Any]
        for k, v in self.get_config_value("inputs_schema").items():
            data = {
                "type": v["type"],
                "doc": v.get("doc", "-- n/a --"),
                "optional": v.get("optional", True),
            }
            result[k] = data

        return result

    def create_outputs_schema(
        self,
    ) -> ValueMapSchema:
        result = {}
        field_name: str
        field_output: MockOutput
        for field_name, field_output in self.get_config_value("outputs").items():
            field_schema = field_output.field_schema
            if field_schema:
                data = {
                    "type": field_schema["type"],
                    "doc": field_schema.get("doc", DEFAULT_NO_DESC_VALUE),
                    "optional": field_schema.get("optional", False),
                }
            else:
                data = {
                    "type": "any",
                    "doc": DEFAULT_NO_DESC_VALUE,
                    "optional": False,
                }
            result[field_name] = data

        return result

    def _retrieve_module_characteristics(self) -> ModuleCharacteristics:
        return ModuleCharacteristics(
            is_idempotent=True, is_internal=True, unique_result_values=True
        )

    def process(self, inputs: ValueMap, outputs: ValueMap) -> None:
        # config = self.get_config_value("desc")

        mock_outputs = self.get_config_value("outputs")
        field_name: str
        field_output: MockOutput
        for field_name, field_output in mock_outputs.items():
            outputs.set_value(field_name, field_output.data)

Attributes

_config_cls = MockModuleConfig class-attribute instance-attribute

Functions

create_inputs_schema() -> ValueMapSchema
Source code in src/kiara/modules/included_core_modules/mock.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def create_inputs_schema(
    self,
) -> ValueMapSchema:
    result = {}
    v: Mapping[str, Any]
    for k, v in self.get_config_value("inputs_schema").items():
        data = {
            "type": v["type"],
            "doc": v.get("doc", "-- n/a --"),
            "optional": v.get("optional", True),
        }
        result[k] = data

    return result
create_outputs_schema() -> ValueMapSchema
Source code in src/kiara/modules/included_core_modules/mock.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def create_outputs_schema(
    self,
) -> ValueMapSchema:
    result = {}
    field_name: str
    field_output: MockOutput
    for field_name, field_output in self.get_config_value("outputs").items():
        field_schema = field_output.field_schema
        if field_schema:
            data = {
                "type": field_schema["type"],
                "doc": field_schema.get("doc", DEFAULT_NO_DESC_VALUE),
                "optional": field_schema.get("optional", False),
            }
        else:
            data = {
                "type": "any",
                "doc": DEFAULT_NO_DESC_VALUE,
                "optional": False,
            }
        result[field_name] = data

    return result
process(inputs: ValueMap, outputs: ValueMap) -> None
Source code in src/kiara/modules/included_core_modules/mock.py 
123
124
125
126
127
128
129
130
def process(self, inputs: ValueMap, outputs: ValueMap) -> None:
    # config = self.get_config_value("desc")

    mock_outputs = self.get_config_value("outputs")
    field_name: str
    field_output: MockOutput
    for field_name, field_output in mock_outputs.items():
        outputs.set_value(field_name, field_output.data)

Functions

default_mock_output() -> Dict[str, MockOutput]

Source code in src/kiara/modules/included_core_modules/mock.py 
19
20
21
22
23
24
25
def default_mock_output() -> Dict[str, MockOutput]:
    schema = {
        "type": "any",
        "doc": "A result",
        "optional": False,
    }
    return {"result": MockOutput(field_schema=schema, data="mock result data")}