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

import logging
import logging
import re
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Iterable, List, Tuple
from uuid import UUID

from django.db import IntegrityError, models

from ferrosoft.apps.dataimport.services.fixtures import (
    load_converters,
    convert_catalog_files,
)
from ferrosoft.apps.ferrobase.models import Tenant

DATAIMPORT_CONVERTERS = (
    "ferrosoft.apps.dataimport.services.conversion.dataimport_converters"
)




def import_json(converters: str, catalog: Iterable[str], schema: str) -> list:
    converters = load_converters(converters)
    fixtures = convert_catalog_files(catalog, schema, converters)
    importer = ObjectCreatingFixtureImporter()
    importer.load_fixtures(fixtures)
    return fixtures





class ImportCallback(ABC):


    @abstractmethod
    def on_object(self, instance):
        raise NotImplementedError




    @abstractmethod
    def on_finish(self):
        return






class FixtureImporter(ABC):


    def load_fixtures(self, fixtures: Iterable[dict], tenant: Tenant | None = None):
        raise NotImplementedError






class ObjectCreatingFixtureImporter(FixtureImporter):
    """
    Importer creating model instances from fixtures.

    This is trimmed down version of Django management command "loaddata".
    """



    @dataclass
    class M2MRelations:
        field_name: str
        related_ids: List[UUID]


    _ID_PATTERN = r"^[0-9a-fA-F]{8}\b-[0-9a-fA-F]{4}\b-[0-9a-fA-F]{4}\b-[0-9a-fA-F]{4}\b-[0-9a-fA-F]{12}$"

    def __init__(self, callback: ImportCallback | None = None):
        self.id_pattern = re.compile(self._ID_PATTERN)
        self._callback = callback



    def load_fixtures(self, fixtures: Iterable[dict], tenant: Tenant | None = None):
        from django.apps import apps

        for fixture in fixtures:
            model_type = apps.get_model(fixture["model"])
            fields, m2m = self._transform_fields(fixture["fields"], model_type)
            model_id = fixture["pk"]

            existing_model = model_type.objects.filter(pk=model_id)
            if existing_model:
                logging.debug("Updating %s %s", model_type, model_id)
                existing_model.update(**fields)
            else:
                logging.debug("Inserting %s %s", model_type, model_id)
                instance = model_type(id=model_id, **fields)
                try:
                    instance.save()
                    # Add any M2M relations after saving.
                    if m2m is not None:
                        m2m_set = getattr(instance, m2m.field_name)
                        for ref in m2m.related_ids:
                            m2m_set.add(ref)
                    if self._callback is not None:
                        self._callback.on_object(instance)
                except IntegrityError:
                    logging.exception("Integrity error while saving instance")

        if self._callback is not None:
            self._callback.on_finish()


    def _transform_fields(self, fields: dict, model_type) -> Tuple[dict, M2MRelations]:
        result = {}
        m2m = None
        for key, value in fields.items():
            # Skip many to many fields. One has to use the set() method.
            field = model_type._meta.get_field(key)
            if isinstance(field, models.ManyToManyField):
                if isinstance(value, list):
                    m2m = self.M2MRelations(key, [UUID(ref) for ref in value])
                continue

            result_key = None
            if isinstance(value, str) and not key.endswith("_id"):
                # If the value matches a UUID, append _id to satisfy Django model constructor for foreign keys.
                match = self.id_pattern.fullmatch(value)
                if match is not None:
                    result_key = "%s_id" % key
            result[result_key or key] = value
        return result, m2m