Skip to content

included_core_operations

Attributes

logger = structlog.getLogger() module-attribute

Classes

CustomModuleOperationDetails

Bases: OperationDetails

Source code in kiara/operations/included_core_operations/__init__.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
56
57
58
59
class CustomModuleOperationDetails(OperationDetails):
    @classmethod
    def create_from_module(cls, module: KiaraModule):

        return CustomModuleOperationDetails(
            operation_id=module.module_type_name,
            module_inputs_schema=module.inputs_schema,
            module_outputs_schema=module.outputs_schema,
        )

    module_inputs_schema: Mapping[str, ValueSchema] = Field(
        description="The input schemas of the module."
    )
    module_outputs_schema: Mapping[str, ValueSchema] = Field(
        description="The output schemas of the module."
    )
    _op_schema: OperationSchema = PrivateAttr(default=None)

    def get_operation_schema(self) -> OperationSchema:

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

        self._op_schema = OperationSchema(
            alias=self.operation_id,
            inputs_schema=self.module_inputs_schema,
            outputs_schema=self.module_outputs_schema,
        )
        return self._op_schema

Attributes

module_inputs_schema: Mapping[str, ValueSchema] = Field(description='The input schemas of the module.') class-attribute
module_outputs_schema: Mapping[str, ValueSchema] = Field(description='The output schemas of the module.') class-attribute

Functions

create_from_module(module: KiaraModule) classmethod
Source code in kiara/operations/included_core_operations/__init__.py 
32
33
34
35
36
37
38
39
@classmethod
def create_from_module(cls, module: KiaraModule):

    return CustomModuleOperationDetails(
        operation_id=module.module_type_name,
        module_inputs_schema=module.inputs_schema,
        module_outputs_schema=module.outputs_schema,
    )
get_operation_schema() -> OperationSchema
Source code in kiara/operations/included_core_operations/__init__.py 
49
50
51
52
53
54
55
56
57
58
59
def get_operation_schema(self) -> OperationSchema:

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

    self._op_schema = OperationSchema(
        alias=self.operation_id,
        inputs_schema=self.module_inputs_schema,
        outputs_schema=self.module_outputs_schema,
    )
    return self._op_schema

CustomModuleOperationType

Bases: OperationType[CustomModuleOperationDetails]

Source code in kiara/operations/included_core_operations/__init__.py 
 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
class CustomModuleOperationType(OperationType[CustomModuleOperationDetails]):

    _operation_type_name = "custom_module"

    def retrieve_included_operation_configs(
        self,
    ) -> Iterable[Union[Mapping, OperationConfig]]:

        result = []
        for name, module_cls in self._kiara.module_type_classes.items():
            mod_conf = module_cls._config_cls
            if mod_conf.requires_config():
                logger.debug(
                    "ignore.custom_operation",
                    module_type=name,
                    reason="config required",
                )
                continue
            doc = DocumentationMetadataModel.from_class_doc(module_cls)
            oc = ManifestOperationConfig(module_type=name, doc=doc)
            result.append(oc)
        return result

    def check_matching_operation(
        self, module: "KiaraModule"
    ) -> Union[CustomModuleOperationDetails, None]:
        mod_conf = module.__class__._config_cls

        if not mod_conf.requires_config():
            is_internal = module.characteristics.is_internal
            # inputs_map = {k: k for k in module.inputs_schema.keys()}
            # outputs_map = {k: k for k in module.outputs_schema.keys()}
            op_details = CustomModuleOperationDetails.create_operation_details(
                operation_id=module.module_type_name,
                module_inputs_schema=module.inputs_schema,
                module_outputs_schema=module.outputs_schema,
                is_internal_operation=is_internal,
            )
            return op_details
        else:
            return None

Functions

retrieve_included_operation_configs() -> Iterable[Union[Mapping, OperationConfig]]
Source code in kiara/operations/included_core_operations/__init__.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def retrieve_included_operation_configs(
    self,
) -> Iterable[Union[Mapping, OperationConfig]]:

    result = []
    for name, module_cls in self._kiara.module_type_classes.items():
        mod_conf = module_cls._config_cls
        if mod_conf.requires_config():
            logger.debug(
                "ignore.custom_operation",
                module_type=name,
                reason="config required",
            )
            continue
        doc = DocumentationMetadataModel.from_class_doc(module_cls)
        oc = ManifestOperationConfig(module_type=name, doc=doc)
        result.append(oc)
    return result
check_matching_operation(module: KiaraModule) -> Union[CustomModuleOperationDetails, None]
Source code in kiara/operations/included_core_operations/__init__.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def check_matching_operation(
    self, module: "KiaraModule"
) -> Union[CustomModuleOperationDetails, None]:
    mod_conf = module.__class__._config_cls

    if not mod_conf.requires_config():
        is_internal = module.characteristics.is_internal
        # inputs_map = {k: k for k in module.inputs_schema.keys()}
        # outputs_map = {k: k for k in module.outputs_schema.keys()}
        op_details = CustomModuleOperationDetails.create_operation_details(
            operation_id=module.module_type_name,
            module_inputs_schema=module.inputs_schema,
            module_outputs_schema=module.outputs_schema,
            is_internal_operation=is_internal,
        )
        return op_details
    else:
        return None