You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
103 lines
2.6 KiB
Python
103 lines
2.6 KiB
Python
"""
|
|
Author: Michel Peltriaux
|
|
Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany
|
|
Contact: michel.peltriaux@sgdnord.rlp.de
|
|
Created on: 21.01.22
|
|
|
|
"""
|
|
from abc import abstractmethod
|
|
|
|
from django.http import JsonResponse
|
|
from django.views import View
|
|
|
|
from api.models import APIUserToken
|
|
from api.settings import KSP_TOKEN_HEADER_IDENTIFIER
|
|
|
|
|
|
class AbstractModelAPIView(View):
|
|
""" Base class for API views
|
|
|
|
The API must follow the GeoJSON Specification RFC 7946
|
|
https://geojson.org/
|
|
https://datatracker.ietf.org/doc/html/rfc7946
|
|
|
|
"""
|
|
model = None
|
|
user = None
|
|
lookup = None
|
|
properties_data = None
|
|
|
|
class Meta:
|
|
abstract = True
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
self.lookup = {
|
|
"id": None, # must be set in subclasses
|
|
"deleted__isnull": True,
|
|
"users__in": [], # must be set in subclasses
|
|
}
|
|
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 fetch_and_serialize(self):
|
|
""" Serializes the model entry according to the given lookup data
|
|
|
|
Args:
|
|
|
|
Returns:
|
|
serialized_data (dict)
|
|
"""
|
|
entry = self.model.objects.get(**self.lookup)
|
|
serialized_data = self.model_to_geo_json(entry)
|
|
return serialized_data
|
|
|
|
def dispatch(self, request, *args, **kwargs):
|
|
try:
|
|
self.user = APIUserToken.get_user_from_token(request.headers.get(KSP_TOKEN_HEADER_IDENTIFIER, None))
|
|
except PermissionError as e:
|
|
return self.return_error_response(e, 403)
|
|
return super().dispatch(request, *args, **kwargs)
|
|
|
|
def return_error_response(self, error, status_code=500):
|
|
""" Returns an error as JsonReponse
|
|
|
|
Args:
|
|
error (): The error/exception
|
|
status_code (): The desired status code
|
|
|
|
Returns:
|
|
|
|
"""
|
|
content = [error.__str__()]
|
|
if hasattr(error, "messages"):
|
|
content = error.messages
|
|
return JsonResponse(
|
|
{
|
|
"errors": content
|
|
},
|
|
status=status_code
|
|
)
|