mpeltriaux
2fa2876090
* adds support for POST of new compensations * adds shared_users property to BaseObject and Compensation to simplify fetching of shared users (Compensation inherits from intervention) * extends compensation admin index * modifies compensation manager which led to invisibility of deleted entries in the admin backend * fixes bug in sanitize_db.py where CREATED useractions would be removed if they are not found on any log but still are used on the .created attribute of the objects
238 lines
8.5 KiB
Python
238 lines
8.5 KiB
Python
"""
|
|
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
|
|
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 intervention 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,
|
|
users__in=[user],
|
|
)
|
|
obj.intervention = intervention
|
|
return obj
|
|
|
|
def set_deadlines(self, obj, deadline_data):
|
|
found_deadlines = []
|
|
for entry in deadline_data:
|
|
deadline_type = entry["type"]
|
|
date = entry["date"]
|
|
comment = entry["comment"]
|
|
|
|
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):
|
|
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):
|
|
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
|