""" 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 django.contrib.gis.geos import MultiPolygon from django.core.exceptions import ObjectDoesNotExist from django.db.models import QuerySet from api.models import ExternalIdentifier from api.utils.serializer.serializer import AbstractModelAPISerializer from codelist.models import KonovaCode from codelist.settings import CODELIST_COMPENSATION_ACTION_ID, CODELIST_BIOTOPES_ID, CODELIST_PROCESS_TYPE_ID, \ CODELIST_LAW_ID, CODELIST_REGISTRATION_OFFICE_ID, CODELIST_CONSERVATION_OFFICE_ID, \ CODELIST_COMPENSATION_ACTION_DETAIL_ID, CODELIST_HANDLER_ID, \ CODELIST_BIOTOPES_EXTRA_CODES_FULL_ID from compensation.models import CompensationAction, UnitChoices, CompensationState from intervention.models import Responsibility, Legal, Handler from konova.models import Deadline, DeadlineType from konova.utils.message_templates import DATA_UNSHARED class AbstractModelAPISerializerV1(AbstractModelAPISerializer): def _model_to_geo_json(self, entry): """ Adds the basic data, which all elements hold Args: entry (): The data entry Returns: """ if entry.geometry.geom is not None: geom = entry.geometry.geom.geojson else: geom = MultiPolygon().geojson geo_json = json.loads(geom) self.properties_data = { "id": entry.id, "identifier": entry.identifier, "title": entry.title, "created_on": self._created_on_to_json(entry), "modified_on": self._modified_on_to_json(entry), } self._extend_properties_data(entry) geo_json["properties"] = self.properties_data return geo_json def _konova_code_to_json(self, konova_code: KonovaCode): """ Serializes KonovaCode model into json Args: konova_code (KonovaCode): The KonovaCode entry Returns: serialized_json (dict) """ if konova_code is None: return None return { "id": konova_code.id, "long_name": konova_code.long_name, "short_name": konova_code.short_name, } def _konova_code_from_json(self, json_str, code_list_identifier): """ Returns a konova code instance Args: json_str (str): The value for the code (id) code_list_identifier (str): From which konova code list this code is supposed to be from Returns: """ if json_str is None: return None json_str = str(json_str) if len(json_str) == 0: return None try: code = KonovaCode.objects.get( id=json_str, code_lists__in=[code_list_identifier] ) except ObjectDoesNotExist as e: msg = f"{e.args[0]} ({json_str} not found in official list {code_list_identifier})" raise ObjectDoesNotExist(msg) return code def _created_on_to_json(self, entry): """ Serializes the created_on into json Args: entry (BaseObject): The entry Returns: created_on (timestamp) """ return entry.created.timestamp if entry.created is not None else None def _modified_on_to_json(self, entry): """ Serializes the modified_on into json Args: entry (BaseObject): The entry Returns: modified_on (timestamp) """ modified_on = entry.modified or entry.created modified_on = modified_on.timestamp if modified_on is not None else None return modified_on def delete_entry(self, id, user): """ Marks an entry as deleted Args: id (str): The entry's id user (User): The API user Returns: """ entry = self._get_obj_from_db(id, user) is_shared = entry.is_shared_with(user) if not is_shared: raise PermissionError(DATA_UNSHARED) # Do not send mails if entry is deleting using API. THere could be hundreds of deletion resulting in hundreds of # mails at once. entry.mark_as_deleted(user, send_mail=False) entry.refresh_from_db() success = entry.deleted is not None return success def _set_external_identifier(self, obj, external_identifier): """ If an external identifier was provided in the payload, we set it in the database Args: obj (BaseObject): The already processed konova object (EIV, KOM, ...) external_identifier (any): The external identifier taken from the payload Returns: """ if external_identifier is None: return None ext_id_obj = ExternalIdentifier.objects.get_or_create( internal_id=obj.id, external_id=external_identifier )[0] if not ext_id_obj.created: ext_id_obj.created = obj.created ext_id_obj.save() return ext_id_obj def _get_external_identifier(self, external_identifier): """ Checks whether a linkage based on an external identifier already exists and returns it if so. Args: external_identifier (any): The external identifier according to payload Returns: ExternalIdentifier | None """ if external_identifier: try: obj = ExternalIdentifier.objects.get(external_id=external_identifier) return obj except ObjectDoesNotExist: pass return None def _check_external_identifier_on_entry_creation(self, external_identifier): """ Special check for POST processing: Checks whether an external identifier already exists on the database. This hints that the entry already has been created in the past. Instead of POST, the PUT method shall be used to avoid creating duplicates. Args: external_identifier (any): The external identifier according to payload Returns: """ persisted_external_identifier = self._get_external_identifier(external_identifier) if persisted_external_identifier: raise AssertionError(f"{external_identifier} has already been initially created! Use PUT for updates!") class DeductableAPISerializerV1Mixin: class Meta: abstract = True def _single_deduction_to_json(self, entry): """ Serializes a single eco account deduction into json Args: entry (EcoAccountDeduction): An EcoAccountDeduction Returns: serialized_json (dict) """ return { "id": entry.pk, "eco_account": { "id": entry.account.pk, "identifier": entry.account.identifier, "title": entry.account.title, }, "surface": entry.surface, "intervention": { "id": entry.intervention.pk, "identifier": entry.intervention.identifier, "title": entry.intervention.title, } } def _deductions_to_json(self, qs: QuerySet): """ Serializes eco account deductions into json Args: qs (QuerySet): A queryset of EcoAccountDeduction entries Returns: serialized_json (list) """ return [ self._single_deduction_to_json(entry) for entry in qs ] class ResponsibilityAPISerializerV1Mixin: class Meta: abstract = True def _handler_to_json(self, handler: Handler): return { "type": self._konova_code_to_json(handler.type), "detail": handler.detail } def _responsible_to_json(self, responsible: Responsibility): """ Serializes Responsibility model into json Args: responsible (Responsibility): The Responsibility entry Returns: serialized_json (dict) """ return { "registration_office": self._konova_code_to_json(responsible.registration_office), "registration_file_number": responsible.registration_file_number, "conservation_office": self._konova_code_to_json(responsible.conservation_office), "conservation_file_number": responsible.conservation_file_number, "handler": self._handler_to_json(responsible.handler), } def _set_responsibility(self, obj, responsibility_data: dict): """ Sets the responsible data contents to the provided responsibility_data dict Args: obj (Intervention): The intervention object responsibility_data (dict): The new data Returns: obj """ if responsibility_data is None: return obj obj.responsible.registration_office = self._konova_code_from_json( responsibility_data["registration_office"], CODELIST_REGISTRATION_OFFICE_ID ) obj.responsible.registration_file_number = responsibility_data["registration_file_number"] obj.responsible.conservation_office = self._konova_code_from_json( responsibility_data["conservation_office"], CODELIST_CONSERVATION_OFFICE_ID, ) obj.responsible.conservation_file_number = responsibility_data["conservation_file_number"] obj.responsible.handler.type = self._konova_code_from_json( responsibility_data["handler"]["type"], CODELIST_HANDLER_ID, ) obj.responsible.handler.detail = responsibility_data["handler"]["detail"] return obj class LegalAPISerializerV1Mixin: class Meta: abstract = True def _legal_to_json(self, legal: Legal): """ Serializes Legal model into json Args: legal (Legal): The Legal entry Returns: serialized_json (dict) """ return { "registration_date": legal.registration_date, "binding_date": legal.binding_date, "process_type": self._konova_code_to_json(legal.process_type), "laws": [self._konova_code_to_json(law) for law in legal.laws.all()], } def _set_legal(self, obj, legal_data): """ Sets the legal data contents to the provided legal_data dict Args: obj (Intervention): The intervention object legal_data (dict): The new data Returns: obj """ if legal_data is None: return obj obj.legal.registration_date = legal_data.get("registration_date", None) obj.legal.binding_date = legal_data.get("binding_date", None) obj.legal.process_type = self._konova_code_from_json( legal_data.get("process_type", None), CODELIST_PROCESS_TYPE_ID, ) laws = [self._konova_code_from_json(law, CODELIST_LAW_ID) for law in legal_data.get("laws", [])] obj.legal.laws.set(laws) return obj class AbstractCompensationAPISerializerV1Mixin: class Meta: abstract = True def _set_deadlines(self, obj, deadline_data): """ Sets the linked deadline data according to the given deadline_data Args: obj (Compensation): The Compensation object deadline_data (dict): The posted deadline_data Returns: obj (Compensation) """ deadlines = [] for entry in deadline_data: try: deadline_type = entry["type"] date = entry["date"] comment = entry["comment"] except KeyError: raise ValueError(f"Invalid deadline content. Content was {entry} but should follow the specification") # Check on validity if deadline_type not in DeadlineType: raise ValueError(f"Invalid deadline type. Choices are {DeadlineType.values}") # If this exact data is already existing, we do not create it new. Instead put it's id in the list of # entries, we will use to set the new actions pre_existing_deadline = obj.deadlines.filter( type=deadline_type, date=date, comment=comment, ).exclude( id__in=deadlines ).first() if pre_existing_deadline is not None: deadlines.append(pre_existing_deadline.id) else: # Create and add id to list new_deadline = Deadline.objects.create( type=deadline_type, date=date, comment=comment, ) deadlines.append(new_deadline.id) obj.deadlines.set(deadlines) return obj def _set_compensation_states(self, obj, states_data, states_manager): """ Sets the linked compensation state data according to the given states_data Args: obj (Compensation): The Compensation object states_data (dict): The posted states_data states_manager (Manager): The before_states or after_states manager Returns: obj (Compensation) """ states = [] for entry in states_data: try: biotope_type = entry["biotope"] biotope_details = [ self._konova_code_from_json(e, CODELIST_BIOTOPES_EXTRA_CODES_FULL_ID) for e in entry["biotope_details"] ] surface = float(entry["surface"]) except KeyError: raise ValueError(f"Invalid biotope content. Content was {entry} but should follow the specification ") # Check on validity if surface <= 0: raise ValueError("State surfaces must be > 0") # If this exact data is already existing, we do not create it new. Instead put it's id in the list of # entries, we will use to set the new actions state = states_manager.filter( biotope_type__id=biotope_type, surface=surface, ).exclude( id__in=states ).first() if state is not None: states.append(state.id) else: # Create and add id to list state = CompensationState.objects.create( biotope_type=self._konova_code_from_json(biotope_type, CODELIST_BIOTOPES_ID), surface=surface ) states.append(state.id) state.biotope_type_details.set(biotope_details) states_manager.set(states) return obj def _set_compensation_actions(self, obj, actions_data): """ Sets the linked compensation action data according to the given actions_data Args: obj (Compensation): The Compensation object actions_data (dict): The posted actions_data Returns: obj (Compensation) """ actions = [] for entry in actions_data: try: action_types = [ self._konova_code_from_json(e, CODELIST_COMPENSATION_ACTION_ID) for e in entry["action_types"] ] action_details = [ self._konova_code_from_json(e, CODELIST_COMPENSATION_ACTION_DETAIL_ID) for e in entry["action_details"] ] amount = float(entry["amount"]) # Mapping of old "qm" into "m²" unit = UnitChoices.m2.value if entry["unit"] == "qm" else entry["unit"] comment = entry["comment"] except KeyError: raise ValueError(f"Invalid action content. Content was {entry} but should follow specification") # Check on validity if amount <= 0: raise ValueError("Action amount must be > 0") if unit not in UnitChoices: raise ValueError(f"Invalid unit. Choices are {UnitChoices.values}") # If this exact data is already existing, we do not create it new. Instead put it's id in the list of # entries, we will use to set the new actions action_entry = obj.actions.filter( action_type__in=action_types, amount=amount, unit=unit, comment=comment, ).exclude( id__in=actions ).first() if action_entry is not None: actions.append(action_entry.id) else: # Create and add id to list action_entry = CompensationAction.objects.create( amount=amount, unit=unit, comment=comment, ) actions.append(action_entry.id) action_entry.action_type.set(action_types) action_entry.action_type_details.set(action_details) obj.actions.set(actions) return obj def _compensation_state_to_json(self, qs: QuerySet): """ Serializes compensation states into json Args: qs (QuerySet): A queryset of CompensationState entries Returns: serialized_json (list) """ return [ { "biotope": self._konova_code_to_json(entry.biotope_type), "biotope_details": [ self._konova_code_to_json(detail) for detail in entry.biotope_type_details.all() ], "surface": entry.surface, } for entry in qs ] def _compensation_actions_to_json(self, qs: QuerySet): """ Serializes CompensationActions into json Args: qs (QuerySet): A queryset of CompensationAction entries Returns: serialized_json (list) """ return [ { "action_types": [ self._konova_code_to_json(action) for action in entry.action_type.all() ], "action_details": [ self._konova_code_to_json(detail) for detail in entry.action_type_details.all() ], "amount": entry.amount, "unit": entry.unit, "comment": entry.comment, } for entry in qs ] def _deadlines_to_json(self, qs: QuerySet): """ Serializes deadlines into json Args: qs (QuerySet): A queryset of Deadline entries Returns: serialized_json (list) """ return list(qs.values( "type", "date", "comment", ))