""" 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.db.models import QuerySet 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_BIOTOPES_EXTRA_CODES_ID, CODELIST_HANDLER_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 { "atom_id": konova_code.atom_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 (atom 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 code = KonovaCode.objects.get( atom_id=json_str, code_lists__in=[code_list_identifier] ) 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 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: deadline_type = entry["type"] date = entry["date"] comment = entry["comment"] # 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: biotope_type = entry["biotope"] biotope_details = [ self._konova_code_from_json(e, CODELIST_BIOTOPES_EXTRA_CODES_ID) for e in entry["biotope_details"] ] surface = float(entry["surface"]) # 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__atom_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: 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"]) unit = entry["unit"] comment = entry["comment"] # 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", ))