Skip to content

metadata

Classes

ExtractMetadataModule (KiaraModule)

Base class to use when writing a module to extract metadata from a file.

It's possible to use any arbitrary kiara module for this purpose, but sub-classing this makes it easier.

Source code in kiara/modules/included_core_modules/metadata.py
class ExtractMetadataModule(KiaraModule):
    """Base class to use when writing a module to extract metadata from a file.

    It's possible to use any arbitrary *kiara* module for this purpose, but sub-classing this makes it easier.
    """

    _config_cls = MetadataModuleConfig
    _module_type_name: str = "value.extract_metadata"

    def _retrieve_module_characteristics(self) -> ModuleCharacteristics:

        return ModuleCharacteristics(
            is_idempotent=True, is_internal=True, unique_result_values=True
        )

    def create_inputs_schema(
        self,
    ) -> Mapping[str, Union[ValueSchema, Mapping[str, Any]]]:

        data_type_name = self.get_config_value("data_type")
        inputs = {
            "value": {
                "type": data_type_name,
                "doc": f"A value of type '{data_type_name}'",
                "optional": False,
            }
        }
        return inputs

    def create_outputs_schema(
        self,
    ) -> Mapping[str, Union[ValueSchema, Mapping[str, Any]]]:

        kiara_model_id: str = self.get_config_value("kiara_model_id")

        # TODO: check it's subclassing the right class

        outputs = {
            "value_metadata": {
                "type": "internal_model",
                "type_config": {"kiara_model_id": kiara_model_id},
                "doc": "The metadata for the provided value.",
            }
        }

        return outputs

    def process(self, inputs: ValueMap, outputs: ValueMap) -> None:

        value = inputs.get_value_obj("value")

        kiara_model_id: str = self.get_config_value("kiara_model_id")

        model_registry = ModelRegistry.instance()
        metadata_model_cls: Type[ValueMetadata] = model_registry.get_model_cls(kiara_model_id=kiara_model_id, required_subclass=ValueMetadata)  # type: ignore

        metadata = metadata_model_cls.create_value_metadata(value=value)

        if not isinstance(metadata, metadata_model_cls):
            raise KiaraProcessingException(
                f"Invalid metadata model result, should be class '{metadata_model_cls.__name__}', but is: {metadata.__class__.__name__}. This is most likely a bug."
            )

        outputs.set_value("value_metadata", metadata)

Classes

_config_cls (KiaraModuleConfig) private pydantic-model
Source code in kiara/modules/included_core_modules/metadata.py
class MetadataModuleConfig(KiaraModuleConfig):

    data_type: str = Field(description="The data type this module will be used for.")
    kiara_model_id: str = Field(description="The id of the kiara (metadata) model.")
Attributes
data_type: str pydantic-field required

The data type this module will be used for.

kiara_model_id: str pydantic-field required

The id of the kiara (metadata) model.

Methods

create_inputs_schema(self)

Return the schema for this types' inputs.

Source code in kiara/modules/included_core_modules/metadata.py
def create_inputs_schema(
    self,
) -> Mapping[str, Union[ValueSchema, Mapping[str, Any]]]:

    data_type_name = self.get_config_value("data_type")
    inputs = {
        "value": {
            "type": data_type_name,
            "doc": f"A value of type '{data_type_name}'",
            "optional": False,
        }
    }
    return inputs
create_outputs_schema(self)

Return the schema for this types' outputs.

Source code in kiara/modules/included_core_modules/metadata.py
def create_outputs_schema(
    self,
) -> Mapping[str, Union[ValueSchema, Mapping[str, Any]]]:

    kiara_model_id: str = self.get_config_value("kiara_model_id")

    # TODO: check it's subclassing the right class

    outputs = {
        "value_metadata": {
            "type": "internal_model",
            "type_config": {"kiara_model_id": kiara_model_id},
            "doc": "The metadata for the provided value.",
        }
    }

    return outputs
process(self, inputs, outputs)
Source code in kiara/modules/included_core_modules/metadata.py
def process(self, inputs: ValueMap, outputs: ValueMap) -> None:

    value = inputs.get_value_obj("value")

    kiara_model_id: str = self.get_config_value("kiara_model_id")

    model_registry = ModelRegistry.instance()
    metadata_model_cls: Type[ValueMetadata] = model_registry.get_model_cls(kiara_model_id=kiara_model_id, required_subclass=ValueMetadata)  # type: ignore

    metadata = metadata_model_cls.create_value_metadata(value=value)

    if not isinstance(metadata, metadata_model_cls):
        raise KiaraProcessingException(
            f"Invalid metadata model result, should be class '{metadata_model_cls.__name__}', but is: {metadata.__class__.__name__}. This is most likely a bug."
        )

    outputs.set_value("value_metadata", metadata)

MetadataModuleConfig (KiaraModuleConfig) pydantic-model

Source code in kiara/modules/included_core_modules/metadata.py
class MetadataModuleConfig(KiaraModuleConfig):

    data_type: str = Field(description="The data type this module will be used for.")
    kiara_model_id: str = Field(description="The id of the kiara (metadata) model.")

Attributes

data_type: str pydantic-field required

The data type this module will be used for.

kiara_model_id: str pydantic-field required

The id of the kiara (metadata) model.