kiara.operations.data_export

DataExportModule

create_input_schema(self)

Abstract method to implement by child classes, returns a description of the input schema of this module.

If returning a dictionary of dictionaries, the format of the return value is as follows (items with '*' are optional):

{ "[input_field_name]: { "type": "[value_type]", "doc*": "[a description of this input]", "optional*': [boolean whether this input is optional or required (defaults to 'False')] "[other_input_field_name]: { "type: ... ... }

Source code in kiara/operations/data_export.py

def create_input_schema(
    self,
) -> typing.Mapping[
    str, typing.Union[ValueSchema, typing.Mapping[str, typing.Any]]
]:

    source_type = self.get_config_value("source_type")
    inputs: typing.Dict[str, typing.Any] = {
        source_type: {
            "type": source_type,
            "doc": f"A value of type '{source_type}'.",
        },
        "base_path": {
            "type": "string",
            "doc": "The directory to export the file(s) to.",
            "optional": True,
        },
        "name": {
            "type": "string",
            "doc": "The (base) name of the exported file(s).",
        },
    }

    return inputs

create_output_schema(self)

Abstract method to implement by child classes, returns a description of the output schema of this module.

If returning a dictionary of dictionaries, the format of the return value is as follows (items with '*' are optional):

{ "[output_field_name]: { "type": "[value_type]", "doc*": "[a description of this output]" "[other_input_field_name]: { "type: ... ... }

Source code in kiara/operations/data_export.py

def create_output_schema(
    self,
) -> typing.Mapping[
    str, typing.Union[ValueSchema, typing.Mapping[str, typing.Any]]
]:

    outputs = {
        "export_details": {
            "type": "dict",
            "doc": "Details about the exported files/folders.",
        }
    }
    return outputs

retrieve_module_profiles(kiara) classmethod

Retrieve a collection of profiles (pre-set module configs) for this kiara module type.

This is used to automatically create generally useful operations (incl. their ids).

Source code in kiara/operations/data_export.py

@classmethod
def retrieve_module_profiles(
    cls, kiara: Kiara
) -> typing.Mapping[str, typing.Union[typing.Mapping[str, typing.Any], Operation]]:

    all_metadata_profiles: typing.Dict[
        str, typing.Dict[str, typing.Dict[str, typing.Any]]
    ] = {}

    sup_type = cls.get_source_value_type()
    if sup_type not in kiara.type_mgmt.value_type_names:
        log_message(
            f"Ignoring data export operation for type '{sup_type}': type not available"
        )
        return {}

    for attr in dir(cls):
        if not attr.startswith("export_as__"):
            continue

        target_profile = attr[11:]

        op_config = {
            "module_type": cls._module_type_id,  # type: ignore
            "module_config": {
                "source_type": sup_type,
                "target_profile": target_profile,
            },
            "doc": f"Export data of type '{sup_type}' as: {target_profile}.",
        }
        all_metadata_profiles[f"export.{sup_type}.as.{target_profile}"] = op_config

    return all_metadata_profiles

DataExportModuleConfig pydantic-model

source_type: str pydantic-field required

The type of the source data that is going to be exported.

target_profile: str pydantic-field required

The name of the target profile. Used to distinguish different target formats for the same data type.

ExportDataOperationType

Export data from kiara.

Operations of this type use internally handled datasets, and export it to the local file system.

Export operations are created by implementing a class that inherits from DataExportModule, kiara will register it under an operation id following this template:

<SOURCE_DATA_TYPE>.export_as.<EXPORT_PROFILE>

The meaning of the templated fields is:

• EXPORTED_DATA_TYPE: the data type of the value to export
• EXPORT_PROFILE: a short, free-form description of the format the data will be exported as

get_export_operations_for_source_type(self, value_type)

Return all available import operations that produce data of the specified type.

Source code in kiara/operations/data_export.py

def get_export_operations_for_source_type(
    self, value_type: str
) -> typing.Dict[str, typing.Dict[str, Operation]]:
    """Return all available import operations that produce data of the specified type."""

    return self.get_export_operations_per_source_type().get(value_type, {})

get_export_operations_per_source_type(self)

Return all available import operations per value type.

The result dictionary uses the source type as first level key, a source name/description as 2nd level key, and the Operation object as value.

Source code in kiara/operations/data_export.py

def get_export_operations_per_source_type(
    self,
) -> typing.Dict[str, typing.Dict[str, typing.Dict[str, Operation]]]:
    """Return all available import operations per value type.

    The result dictionary uses the source type as first level key, a source name/description as 2nd level key,
    and the Operation object as value.
    """

    result: typing.Dict[str, typing.Dict[str, typing.Dict[str, Operation]]] = {}

    for op_config in self.operations.values():

        target_type: str = op_config.module_cls.get_source_value_type()  # type: ignore

        source_type = op_config.module_config["source_type"]
        target_profile = op_config.module_config["target_profile"]

        result.setdefault(target_type, {}).setdefault(source_type, {})[
            target_profile
        ] = op_config

    return result

FileBundleImportModule

Import a file, optionally saving it to the data store.

FileExportModule

Import a file, optionally saving it to the data store.