Skip to content

metadata

Classes

ExtractMetadataDetails

Bases: BaseOperationDetails

A model that contains information needed to describe an 'extract_metadata' operation.

Source code in kiara/operations/included_core_operations/metadata.py 
23
24
25
26
27
28
29
30
31
class ExtractMetadataDetails(BaseOperationDetails):
    """A model that contains information needed to describe an 'extract_metadata' operation."""

    data_type: str = Field(
        description="The data type this metadata operation can be used with."
    )
    metadata_key: str = Field(description="The metadata key.")
    input_field_name: str = Field(description="The input field name.")
    result_field_name: str = Field(description="The result field name.")

Attributes

data_type: str = Field(description='The data type this metadata operation can be used with.') class-attribute
metadata_key: str = Field(description='The metadata key.') class-attribute
input_field_name: str = Field(description='The input field name.') class-attribute
result_field_name: str = Field(description='The result field name.') class-attribute

ExtractMetadataOperationType

Bases: OperationType[ExtractMetadataDetails]

An operation that extracts metadata of a specific type from value data.

For a module profile to be picked up by this operation type, it needs to have: - exactly one input field - that input field must have the same name as its value type, or be 'value' - exactly one output field, whose field name is called 'value_metadata', and where the value has the type 'internal_model'

Source code in kiara/operations/included_core_operations/metadata.py 
 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
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
class ExtractMetadataOperationType(OperationType[ExtractMetadataDetails]):
    """An operation that extracts metadata of a specific type from value data.

    For a module profile to be picked up by this operation type, it needs to have:
    - exactly one input field
    - that input field must have the same name as its value type, or be 'value'
    - exactly one output field, whose field name is called 'value_metadata', and where the value has the type 'internal_model'
    """

    _operation_type_name = "extract_metadata"

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

        model_registry = ModelRegistry.instance()
        all_models = model_registry.get_models_of_type(ValueMetadata)

        result = []
        for model_id, model_cls_info in all_models.item_infos.items():
            model_cls: Type[ValueMetadata] = model_cls_info.python_class.get_class()  # type: ignore
            metadata_key = model_cls._metadata_key  # type: ignore
            data_types = model_cls.retrieve_supported_data_types()
            if isinstance(data_types, str):
                data_types = [data_types]
            for data_type in data_types:

                config = {
                    "module_type": "value.extract_metadata",
                    "module_config": {
                        "data_type": data_type,
                        "kiara_model_id": model_cls._kiara_model_id,  # type: ignore
                    },
                    "doc": f"Extract '{metadata_key}' metadata for value type '{data_type}'.",
                }
                result.append(config)

        return result

    def check_matching_operation(
        self, module: "KiaraModule"
    ) -> Union[ExtractMetadataDetails, None]:

        if len(module.outputs_schema) != 1:
            return None
        if (
            "value_metadata" not in module.outputs_schema
            or module.outputs_schema["value_metadata"].type != "internal_model"
        ):
            return None
        if len(module.inputs_schema) != 1:
            return None

        input_field_name = next(iter(module.inputs_schema.keys()))
        input_schema = module.inputs_schema.get(input_field_name)
        assert input_schema is not None
        if input_field_name != input_schema.type and input_field_name != "value":
            return None

        data_type_name = module.inputs_schema["value"].type
        model_id: str = module.get_config_value("kiara_model_id")

        registry = ModelRegistry.instance()
        metadata_model_cls = registry.get_model_cls(
            kiara_model_id=model_id, required_subclass=ValueMetadata
        )

        metadata_key = metadata_model_cls._metadata_key  # type: ignore

        if data_type_name == "any":
            op_id = f"extract.{metadata_key}.metadata"
        else:
            op_id = f"extract.{metadata_key}.metadata.from.{data_type_name}"

        details = ExtractMetadataDetails.create_operation_details(
            module_inputs_schema=module.inputs_schema,
            module_outputs_schema=module.outputs_schema,
            operation_id=op_id,
            data_type=data_type_name,
            metadata_key=metadata_key,
            input_field_name=input_field_name,
            result_field_name="value_metadata",
            is_internal_operation=True,
        )

        return details

    def get_operations_for_data_type(self, data_type: str) -> Mapping[str, Operation]:
        """Return all available metadata extract operations for the provided type (and it's parent types).

        Arguments:
            data_type: the value type

        Returns:
            a mapping with the metadata type as key, and the operation as value
        """

        lineage = set(
            self._kiara.type_registry.get_type_lineage(data_type_name=data_type)
        )

        result = {}

        for op_id, op in self.operations.items():
            op_details = self.retrieve_operation_details(op)
            included = op_details.data_type in lineage
            if not included:
                continue
            metadata_key = op_details.metadata_key
            if metadata_key in result:
                raise Exception(
                    f"Duplicate metadata operations for type '{metadata_key}'."
                )

            result[metadata_key] = op

        return result

Functions

retrieve_included_operation_configs() -> Iterable[Union[Mapping, OperationConfig]]
Source code in kiara/operations/included_core_operations/metadata.py 
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
def retrieve_included_operation_configs(
    self,
) -> Iterable[Union[Mapping, OperationConfig]]:

    model_registry = ModelRegistry.instance()
    all_models = model_registry.get_models_of_type(ValueMetadata)

    result = []
    for model_id, model_cls_info in all_models.item_infos.items():
        model_cls: Type[ValueMetadata] = model_cls_info.python_class.get_class()  # type: ignore
        metadata_key = model_cls._metadata_key  # type: ignore
        data_types = model_cls.retrieve_supported_data_types()
        if isinstance(data_types, str):
            data_types = [data_types]
        for data_type in data_types:

            config = {
                "module_type": "value.extract_metadata",
                "module_config": {
                    "data_type": data_type,
                    "kiara_model_id": model_cls._kiara_model_id,  # type: ignore
                },
                "doc": f"Extract '{metadata_key}' metadata for value type '{data_type}'.",
            }
            result.append(config)

    return result
check_matching_operation(module: KiaraModule) -> Union[ExtractMetadataDetails, None]
Source code in kiara/operations/included_core_operations/metadata.py 
 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
131
132
133
134
135
136
137
138
def check_matching_operation(
    self, module: "KiaraModule"
) -> Union[ExtractMetadataDetails, None]:

    if len(module.outputs_schema) != 1:
        return None
    if (
        "value_metadata" not in module.outputs_schema
        or module.outputs_schema["value_metadata"].type != "internal_model"
    ):
        return None
    if len(module.inputs_schema) != 1:
        return None

    input_field_name = next(iter(module.inputs_schema.keys()))
    input_schema = module.inputs_schema.get(input_field_name)
    assert input_schema is not None
    if input_field_name != input_schema.type and input_field_name != "value":
        return None

    data_type_name = module.inputs_schema["value"].type
    model_id: str = module.get_config_value("kiara_model_id")

    registry = ModelRegistry.instance()
    metadata_model_cls = registry.get_model_cls(
        kiara_model_id=model_id, required_subclass=ValueMetadata
    )

    metadata_key = metadata_model_cls._metadata_key  # type: ignore

    if data_type_name == "any":
        op_id = f"extract.{metadata_key}.metadata"
    else:
        op_id = f"extract.{metadata_key}.metadata.from.{data_type_name}"

    details = ExtractMetadataDetails.create_operation_details(
        module_inputs_schema=module.inputs_schema,
        module_outputs_schema=module.outputs_schema,
        operation_id=op_id,
        data_type=data_type_name,
        metadata_key=metadata_key,
        input_field_name=input_field_name,
        result_field_name="value_metadata",
        is_internal_operation=True,
    )

    return details
get_operations_for_data_type(data_type: str) -> Mapping[str, Operation]

Return all available metadata extract operations for the provided type (and it's parent types).

Parameters:

Name Type Description Default
data_type str

the value type

 required

Returns:

Type Description
Mapping[str, Operation]

a mapping with the metadata type as key, and the operation as value
Source code in kiara/operations/included_core_operations/metadata.py 
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
def get_operations_for_data_type(self, data_type: str) -> Mapping[str, Operation]:
    """Return all available metadata extract operations for the provided type (and it's parent types).

    Arguments:
        data_type: the value type

    Returns:
        a mapping with the metadata type as key, and the operation as value
    """

    lineage = set(
        self._kiara.type_registry.get_type_lineage(data_type_name=data_type)
    )

    result = {}

    for op_id, op in self.operations.items():
        op_details = self.retrieve_operation_details(op)
        included = op_details.data_type in lineage
        if not included:
            continue
        metadata_key = op_details.metadata_key
        if metadata_key in result:
            raise Exception(
                f"Duplicate metadata operations for type '{metadata_key}'."
            )

        result[metadata_key] = op

    return result