konova/api/utils/serializer/serializer.py

"""
Author: Michel Peltriaux
Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany
Contact: michel.peltriaux@sgdnord.rlp.de
Created on: 24.01.22

"""
import json
from abc import abstractmethod

from django.contrib.gis import geos
from django.contrib.gis.geos import GEOSGeometry
from django.core.exceptions import ObjectDoesNotExist
from django.core.paginator import Paginator
from django.db.models import Q

from api.models import ExternalIdentifier
from konova.models import Geometry
from konova.utils.message_templates import DATA_UNSHARED


class AbstractModelAPISerializer:
    model = None
    lookup = None
    properties_data = None

    rpp = None
    page_number = None
    paginator = None

    class Meta:
        abstract = True

    def __init__(self, *args, **kwargs):
        self.lookup = {
            "id": None,  # must be set
            "deleted__isnull": True,
        }
        self.shared_lookup = Q()  # must be set, so user or team based share will be used properly
        super().__init__(*args, **kwargs)

    @abstractmethod
    def _model_to_geo_json(self, entry):
        """ Defines the model as geo json

        Args:
            entry (): The found entry from the database

        Returns:

        """
        raise NotImplementedError("Must be implemented in subclasses")

    @abstractmethod
    def _extend_properties_data(self, entry):
        """ Defines the 'properties' part of geo json

        Args:
            entry (): The found entry from the database

        Returns:

        """
        raise NotImplementedError("Must be implemented in subclasses")

    def prepare_lookup(self, _id, user):
        """ Updates lookup dict for db fetching

        Args:
            _id (str): The object's id
            user (User): The user requesting for

        Returns:

        """
        if _id is None:
            # Return all objects
            del self.lookup["id"]
        else:
            # Return certain object
            ## But first check, whether this is an external identifier ...
            try:
                ## If we can find this _id on our ExternalIdentifier model, we need to map it on the internal id
                ext_id = ExternalIdentifier.objects.get(external_id=_id)
                _id = ext_id.internal_id
            except ObjectDoesNotExist:
                # If we did not find it, we assume that this is already an internal id. (Or it does not exist at all)
                pass
            self.lookup["id"] = _id

        self.shared_lookup = Q(
            Q(users__in=[user]) |
            Q(teams__in=list(user.shared_teams))
        )

    def fetch_and_serialize(self):
        """ Serializes the model entry according to the given lookup data

        Args:

        Returns:
            serialized_data (dict)
        """
        entries = self.model.objects.filter(
            **self.lookup
        ).filter(
            self.shared_lookup
        ).order_by(
            "id"
        ).distinct()
        self.paginator = Paginator(entries, self.rpp)
        requested_entries = self.paginator.page(self.page_number)

        serialized_data = {}
        for entry in requested_entries.object_list:
            serialized_data[str(entry.id)] = self._model_to_geo_json(entry)
        return serialized_data

    @abstractmethod
    def update_model_from_json(self, id, json_model, user):
        """ Updates an instance from given json data

        Args:
            id (str): The instance's to be updated
            json_model (dict): JSON data
            user (User): The performing user

        Returns:

        """
        raise NotImplementedError("Must be implemented in subclasses")

    @abstractmethod
    def create_model_from_json(self, json_model, user):
        """ Creates a new instance from given json data

        Args:
            json_model (dict): JSON data
            user (User): The performing user

        Returns:

        """
        raise NotImplementedError("Must be implemented in subclasses")

    def _create_geometry_from_json(self, geojson) -> GEOSGeometry:
        """ Creates a GEOSGeometry object based on the given geojson

        Args:
            geojson (str|dict): The geojson as str or dict

        Returns:
            geometry (GEOSGeometry)
        """
        if isinstance(geojson, dict):
            geojson = json.dumps(geojson)
        geometry = geos.fromstr(geojson)
        geometry = Geometry.cast_to_rlp_srid(geometry)
        geometry = Geometry.cast_to_multipolygon(geometry)
        return geometry

    def _get_obj_from_db(self, id, user):
        """ Returns the object from database

        Fails if id not found or user does not have shared access

        Args:
            id (str): The object's id
            user (User): The API user

        Returns:

        """
        # First if there is an external identifier linked to an internal one, so we can continue with the internal
        try:
            ext_id = ExternalIdentifier.objects.get(external_id=id)
            id = ext_id.internal_id
        except ObjectDoesNotExist:
            # No external id found - let's hope the given id exists internally
            pass

        obj = self.model.objects.get(
            id=id,
            deleted__isnull=True,
        )
        is_shared = obj.is_shared_with(user)
        if not is_shared:
            raise PermissionError(DATA_UNSHARED)
        return obj

    @abstractmethod
    def _initialize_objects(self, json_model, user):
        """ Initializes all needed objects from the json_model data

        Does not persist data to the DB!

        Args:
            json_model (dict): The json data
            user (User): The API user

        Returns:
            obj (Intervention)
        """
        raise NotImplementedError("Must be implemented in subclasses")


    def _geometry_conflicts_to_list(self, geometry) -> list:
        """ Serializes geometry conflict ids into dict

        Args:
            geometry (Geometry): The geometry to fetch geometry conflicts from

        Returns:
            ids (list): Serialized geometry conflicts as dict objects inside a list
        """
        ids = []
        conflict_geometries = geometry.get_conflict_geometries()
        for geom in conflict_geometries:
            try:
                data = geom.get_data_objects(["identifier", "id"])[0]
            except KeyError:
                raise AssertionError(f"Geometry {geom.id} is not attached to any entries. Contact an admin!")
            ids.append(
                {
                    "identifier": data["identifier"],
                    "id": data["id"],
                }
            )

        return ids