Source code for ferrosoft.apps.dataimport.services.conversion

from typing import Iterable

from ferrosoft.apps.dataimport.models import FieldType
from ferrosoft.apps.dataimport.services.errors import CatalogConversionError
from ferrosoft.apps.dataimport.services.fixtures import SchemaID




def convert_enum(enum_type, external_value: str):
    if hasattr(enum_type, external_value):
        return enum_type[external_value]
    raise CatalogConversionError(
        "Invalid %s: %s" % (enum_type.__name__, external_value)
    )





def convert_dataimport_catalog(catalog: dict) -> Iterable[dict]:
    fixtures = []

    for collection in catalog["mappingCollections"]:
        fixtures.append(
            {
                "model": "dataimport.mappingcollection",
                "pk": collection["id"],
                "fields": {
                    "name": collection["name"],
                    "model_name": collection["model"],
                    "documentation": collection.get("documentation", ""),
                },
            }
        )

        for field in collection["fields"]:
            fixtures.append(
                {
                    "model": "dataimport.fieldmapping",
                    "pk": field["id"],
                    "fields": {
                        "collection": collection["id"],
                        "field_type": convert_enum(FieldType, field["type"]),
                        "input_field_name": field["inputField"],
                        "output_field_name": field["modelField"],
                        "related_model": field.get("relatedModel", None),
                        "related_field": field.get("relatedField", None),
                        "required": field["required"],
                        "enum_class": field.get("enumClass", None),
                        "scope": field.get("scope", ""),
                    },
                },
            )

    for jobTemplate in catalog["jobTemplates"]:
        fixtures.append(
            {
                "model": "dataimport.importjobtemplate",
                "pk": jobTemplate["id"],
                "fields": {
                    "name": jobTemplate["name"],
                    "code_name": jobTemplate["codeName"],
                    "collection": jobTemplate["collection"],
                    "text_encoding": jobTemplate["textEncoding"],
                    "processor_before_job": jobTemplate["processorBeforeJob"],
                    "processor_after_job": jobTemplate["processorAfterJob"],
                    "model_processor": jobTemplate["modelProcessor"],
                },
            }
        )

    return fixtures



dataimport_converters = {
    SchemaID(
        "https://emiflow.de/dataimport/dataimport.schema.json"
    ): convert_dataimport_catalog,
}