""" Author: Michel Peltriaux Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany Contact: michel.peltriaux@sgdnord.rlp.de Created on: 24.01.22 """ from django.db import transaction from api.utils.serializer.v1.serializer import AbstractModelAPISerializerV1 from codelist.settings import CODELIST_COMPENSATION_ACTION_ID, CODELIST_BIOTOPES_ID from compensation.models import Compensation, CompensationAction, CompensationState, UnitChoices from intervention.models import Intervention from konova.models import Geometry, Deadline, DeadlineType from konova.tasks import celery_update_parcels from user.models import UserActionLogEntry class CompensationAPISerializerV1(AbstractModelAPISerializerV1): model = Compensation def prepare_lookup(self, id, user): self.lookup["id"] = id del self.lookup["users__in"] self.lookup["intervention__users__in"] = [user] def intervention_to_json(self, entry): return { "id": entry.pk, "identifier": entry.identifier, "title": entry.title, } def extend_properties_data(self, entry): self.properties_data["is_cef"] = entry.is_cef self.properties_data["is_coherence_keeping"] = entry.is_coherence_keeping self.properties_data["intervention"] = self.intervention_to_json(entry.intervention) self.properties_data["before_states"] = self.compensation_state_to_json(entry.before_states.all()) self.properties_data["after_states"] = self.compensation_state_to_json(entry.after_states.all()) self.properties_data["actions"] = self.compensation_actions_to_json(entry.actions.all()) self.properties_data["deadlines"] = self.deadlines_to_json(entry.deadlines.all()) 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 (Compensation) """ create_action = UserActionLogEntry.get_created_action(user, comment="API Import") # Create geometry json_geom = self.create_geometry_from_json(json_model) geometry = Geometry() geometry.geom = json_geom geometry.created = create_action # Create linked objects obj = Compensation() created = create_action obj.created = created obj.geometry = geometry return obj def set_intervention(self, obj, intervention_id, user): """ Sets the linked compensation according to the given id Fails if no such intervention found or user has no shared access Args: obj (Compensation): The Compensation object intervention_id (str): The intervention's id user (User): The API user Returns: obj (Compensation) """ intervention = Intervention.objects.get( id=intervention_id, ) is_shared = intervention.is_shared_with(user) if not is_shared: raise PermissionError("Intervention not shared with user") obj.intervention = intervention return obj 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) """ found_deadlines = [] for entry in deadline_data: deadline_type = entry["type"] date = entry["date"] comment = entry["comment"] if deadline_type not in DeadlineType: raise ValueError(f"Invalid deadline type. Choices are {DeadlineType.values}") pre_existing_deadlines = obj.deadlines.filter( type=deadline_type, date=date, comment=comment, ).exclude( id__in=found_deadlines ) if pre_existing_deadlines.count() > 0: found_deadlines += pre_existing_deadlines.values_list("id", flat=True) else: # Create! new_deadline = Deadline.objects.create( type=deadline_type, date=date, comment=comment, ) obj.deadlines.add(new_deadline) 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) """ found_states = [] for entry in states_data: biotope_type = entry["biotope"] surface = float(entry["surface"]) if surface <= 0: raise ValueError("State surfaces must be > 0") pre_existing_states = states_manager.filter( biotope_type__atom_id=biotope_type, surface=surface, ).exclude( id__in=found_states ) if pre_existing_states.count() > 0: found_states += pre_existing_states.values_list("id", flat=True) else: # Create! new_state = CompensationState.objects.create( biotope_type=self.konova_code_from_json(biotope_type, CODELIST_BIOTOPES_ID), surface=surface ) states_manager.add(new_state) 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) """ found_actions = [] for entry in actions_data: action = entry["action"] amount = float(entry["amount"]) unit = entry["unit"] comment = entry["comment"] if amount <= 0: raise ValueError("Action amount must be > 0") if unit not in UnitChoices: raise ValueError(f"Invalid unit. Choices are {UnitChoices.values}") pre_existing_actions = obj.actions.filter( action_type__atom_id=action, amount=amount, unit=unit, comment=comment, ).exclude( id__in=found_actions ) if pre_existing_actions.count() > 0: found_actions += pre_existing_actions.values_list("id", flat=True) else: # Create! new_action = CompensationAction.objects.create( action_type=self.konova_code_from_json(action, CODELIST_COMPENSATION_ACTION_ID), amount=amount, unit=unit, comment=comment, ) obj.actions.add(new_action) return obj def create_model_from_json(self, json_model, user): """ Creates a new entry for the model based on the contents of json_model Args: json_model (dict): The json containing data user (User): The API user Returns: created_id (str): The id of the newly created Compensation entry """ with transaction.atomic(): obj = self.initialize_objects(json_model, user) # Fill in data to objects properties = json_model["properties"] obj.identifier = obj.generate_new_identifier() obj.title = properties["title"] obj.is_cef = properties["is_cef"] obj.is_coherence_keeping = properties["is_coherence_keeping"] obj = self.set_intervention(obj, properties["intervention"], user) obj.geometry.save() obj.save() obj = self.set_compensation_actions(obj, properties["actions"]) obj = self.set_compensation_states(obj, properties["before_states"], obj.before_states) obj = self.set_compensation_states(obj, properties["after_states"], obj.after_states) obj = self.set_deadlines(obj, properties["deadlines"]) obj.log.add(obj.created) celery_update_parcels.delay(obj.geometry.id) return obj.id def update_model_from_json(self, id, json_model, user): """ Updates an entry for the model based on the contents of json_model Args: id (str): The object's id json_model (dict): The json containing data user (User): The API user Returns: created_id (str): The id of the newly created Compensation entry """ with transaction.atomic(): obj = self.get_obj_from_db(id, user) # Fill in data to objects properties = json_model["properties"] obj.title = properties["title"] self.set_responsibility(obj, properties["responsible"]) self.set_legal(obj, properties["legal"]) obj.geometry.geom = self.create_geometry_from_json(json_model) obj.responsible.save() obj.geometry.save() obj.legal.save() obj.save() obj.users.add(user) celery_update_parcels.delay(obj.geometry.id) return obj.id