Terralego/django-terra-geocrud

View on GitHub
terra_geocrud/serializers.py

Summary

Maintainability
A
50 mins
Test Coverage
import json
from collections import OrderedDict
from copy import deepcopy
from pathlib import Path

from django.template.defaultfilters import date
from django.core.exceptions import ObjectDoesNotExist
from django.utils.module_loading import import_string
from django.utils.text import slugify
from django.utils.translation import gettext_lazy as _
from geostore import settings as geostore_settings
from geostore.models import Feature, LayerExtraGeom
from geostore.serializers import FeatureSerializer, FeatureExtraGeomSerializer, GeometryFileAsyncSerializer
from rest_framework import serializers
from rest_framework.reverse import reverse
from rest_framework_gis import serializers as geo_serializers
from template_model.models import Template

from . import models
from .map.styles import get_default_style
from .properties.files import store_feature_files
from .properties.utils import serialize_group_properties

# use base serializer as defined in geostore settings. using django-geostore-routing change this value

LayerSerializer = import_string(geostore_settings.GEOSTORE_LAYER_SERIALIZER)


class BaseUpdatableMixin(serializers.ModelSerializer):
    created_at = serializers.SerializerMethodField()
    updated_at = serializers.SerializerMethodField()

    def get_created_at(self, obj):
        return date(obj.created_at, 'SHORT_DATETIME_FORMAT')

    def get_updated_at(self, obj):
        return date(obj.updated_at, 'SHORT_DATETIME_FORMAT')


class LayerViewSerializer(LayerSerializer):
    schema = serializers.JSONField(write_only=True)
    # disable other fields
    name = serializers.CharField(read_only=True)
    layer_groups = None
    routing_url = None
    shapefile_url = None
    geojson_url = None
    layer_intersects = None

    class Meta(LayerSerializer.Meta):
        pass


class CrudViewSerializer(serializers.ModelSerializer):
    layer = LayerViewSerializer()
    object_name = serializers.SerializerMethodField()
    object_name_plural = serializers.SerializerMethodField()
    extent = serializers.SerializerMethodField()
    exports = serializers.SerializerMethodField()
    ui_schema = serializers.JSONField(source='grouped_ui_schema')
    form_schema = serializers.JSONField(source='grouped_form_schema')
    map_style = serializers.JSONField(source='map_style_with_default')
    map_layers = serializers.SerializerMethodField(help_text=_("VT styles and definitions"))
    feature_endpoint = serializers.SerializerMethodField(
        help_text=_("Url endpoint for view's features")
    )
    feature_list_properties = serializers.SerializerMethodField(
        help_text=_("Available properties for feature datatable. Ordered, {name: {title, type}}")
    )
    routing_settings = serializers.SerializerMethodField()

    def get_object_name(self, obj):
        return obj.object_name if obj.object_name else obj.name

    def get_object_name_plural(self, obj):
        return obj.object_name_plural if obj.object_name_plural else obj.name

    def get_map_layers(self, obj):
        data = [{
            'title': obj.name,
            # TODO: Add slugify to be consistent with relations
            'id_layer_vt': obj.layer.name,
            'style': obj.map_style_with_default,
            'main': True
        }]

        for relation in obj.layer.relations_as_origin.all():
            layer = relation.destination
            try:
                related_crud_view = layer.crud_view
                view = {
                    'title': related_crud_view.name,
                    'style': related_crud_view.map_style,
                    'id_layer_vt': f'relation-{slugify(obj.layer.name)}-{slugify(relation.name)}',
                    'main': False,
                    'view_source': 'relation',
                    'pk': layer.pk
                }
                data.append(view)
            except ObjectDoesNotExist:
                pass

        # add extra_layer styles
        for extra_layer in obj.layer.extra_geometries.all():
            # get final style
            try:
                style = extra_layer.style.map_style_with_default
            except LayerExtraGeom.style.RelatedObjectDoesNotExist:
                style = get_default_style(extra_layer)

            data.append({
                'title': extra_layer.title,
                # TODO: Add slugify to be consistent with relations
                'id_layer_vt': extra_layer.name,
                'style': style,
                'main': False,
                'view_source': 'extra_geometry',
                'pk': extra_layer.pk
            })
        return data

    def get_exports(self, obj):
        if not geostore_settings.GEOSTORE_EXPORT_CELERY_ASYNC:
            return None
        serializer = GeometryFileAsyncSerializer(obj.layer)
        return serializer.data

    def get_extent(self, obj):
        # TODO: use annotated extent
        return obj.extent

    def get_feature_list_properties(self, obj):
        # TODO: keep default properties at first, then order by property title
        default_list = list(obj.default_list_properties.values_list('key', flat=True)) or list(
            obj.list_available_properties.values_list('key', flat=True))[:8]
        result = {
            prop.key: {
                "title": obj.layer.get_property_title(prop.key),
                "selected": True if prop.key in default_list else False,
                "type": obj.layer.get_property_type(prop.key),
                "table_order": prop.table_order
            }
            for prop in obj.list_available_properties.all()
        }
        # order by title
        return OrderedDict(sorted(result.items(), key=lambda x: x[1]['title']))

    def get_feature_endpoint(self, obj):
        return reverse('feature-list', args=(obj.layer_id,))

    def get_routing_settings(self, obj):
        data = []
        for routing_setting in obj.routing_settings.all():
            label = routing_setting.label
            options = {}
            if routing_setting.provider == "mapbox":
                options["transit"] = routing_setting.mapbox_transit
            else:
                options["url"] = reverse('layer-route', args=[routing_setting.layer.pk])
            data.append({"label": label,
                         "provider": {
                             "name": routing_setting.provider,
                             "options": options
                         }
                         })
        return data

    class Meta:
        model = models.CrudView
        fields = (
            'id', 'name', 'object_name', 'object_name_plural',
            'pictogram', 'order', 'map_style',
            'form_schema', 'ui_schema', 'settings', 'layer',
            'feature_endpoint', 'extent', 'exports',
            'feature_list_properties', 'map_layers',
            'routing_settings'
        )


class CrudGroupSerializer(serializers.ModelSerializer):
    crud_views = CrudViewSerializer(many=True, read_only=True)

    class Meta:
        model = models.CrudGroupView
        fields = '__all__'


class FeatureDisplayPropertyGroup(serializers.ModelSerializer):
    title = serializers.CharField(source='label')
    order = serializers.IntegerField()
    pictogram = serializers.ImageField()
    properties = serializers.SerializerMethodField()

    def get_properties(self, obj):
        """ Get feature properties in group to form { title: rendering(value) } """
        feature = self.context.get('feature')
        final_properties = {
            prop.key: feature.properties.get(prop.key)
            for prop in obj.group_properties.all()
        }
        editable = {prop.key: prop.editable for prop in obj.group_properties.all()}
        return serialize_group_properties(feature, final_properties, editable)

    class Meta:
        model = models.FeaturePropertyDisplayGroup
        fields = ('title', 'slug', 'order', 'pictogram', 'properties')


class CrudFeatureListSerializer(BaseUpdatableMixin, FeatureSerializer):
    geom = None
    detail_url = serializers.SerializerMethodField()
    extent = serializers.SerializerMethodField()
    properties = serializers.SerializerMethodField()

    def get_properties(self, obj):
        """ Keep only properties that can be shown in list """
        keys = list(obj.layer.crud_view.list_available_properties.values_list('key', flat=True))
        return {
            key: value for key, value in obj.properties.items() if key in keys
        }

    def get_extent(self, obj):
        geom = obj.geom.transform(4326, clone=True)
        return geom.extent

    def get_detail_url(self, obj):
        return reverse('feature-detail',
                       args=(obj.layer_id, obj.identifier))

    def get_relations(self, obj):
        return {
            relation.name: reverse('feature-relation',
                                   args=(obj.layer_id, obj.identifier, relation.pk))
            for relation in obj.layer.relations_as_origin.all()
            if hasattr(relation.destination, 'crud_view')
        }

    class Meta(FeatureSerializer.Meta):
        exclude = ('source', 'target', 'layer', 'geom')
        fields = None


class DocumentFeatureSerializer(serializers.ModelSerializer):
    extension = serializers.SerializerMethodField()
    template_name = serializers.CharField(source='name')
    template_file = serializers.SerializerMethodField()
    download_url = serializers.SerializerMethodField()

    def get_extension(self, obj):
        return Path(obj.template_file.name).suffix

    def get_template_file(self, obj):
        return Path(obj.template_file.name).name

    def get_download_url(self, obj):
        feature = self.context.get('feature')
        return reverse('feature-generate-template',
                       args=(feature.layer.pk, feature.identifier, obj.pk))

    class Meta:
        fields = (
            'extension', 'template_name', 'template_file', 'download_url'
        )
        model = Template


class FeaturePictureSerializer(BaseUpdatableMixin):
    thumbnail = serializers.ImageField(read_only=True)
    action_url = serializers.SerializerMethodField()

    def get_action_url(self, obj):
        return reverse('picture-detail', args=(obj.feature.identifier,
                                               obj.pk, ))

    class Meta:
        model = models.FeaturePicture
        extra_kwargs = {
        }
        fields = ('id', 'category', 'legend', 'image', 'thumbnail', 'action_url', 'created_at', 'updated_at')


class FeatureAttachmentSerializer(BaseUpdatableMixin):
    action_url = serializers.SerializerMethodField()

    def get_action_url(self, obj):
        return reverse('attachment-detail', args=(obj.feature.identifier,
                                                  obj.pk, ))

    class Meta:
        model = models.FeatureAttachment
        fields = ('id', 'category', 'legend', 'file', 'action_url', 'created_at', 'updated_at')


class AttachmentCategorySerializer(serializers.ModelSerializer):
    class Meta:
        model = models.AttachmentCategory
        fields = '__all__'


class CrudFeatureDetailSerializer(BaseUpdatableMixin, FeatureSerializer):
    title = serializers.SerializerMethodField()
    geom = geo_serializers.GeometryField()
    documents = serializers.SerializerMethodField()
    display_properties = serializers.SerializerMethodField()
    properties = serializers.JSONField()
    attachments = serializers.SerializerMethodField()
    pictures = serializers.SerializerMethodField()
    geometries = serializers.SerializerMethodField()
    routing_information = serializers.JSONField(source='routing_information.route_description', required=False)

    def create(self, validated_data):
        routing_information = validated_data.pop('routing_information', {})

        feature = Feature.objects.create(**validated_data)
        models.RoutingInformations.objects.create(feature=feature,
                                                  route_description=routing_information.get('route_description',
                                                                                            {}))
        return feature

    def get_update_fields(self, instance, validated_data):
        geom = validated_data.get("geom")
        properties = validated_data.get("properties")
        update_fields = ["updated_at"]
        if geom and geom != instance.geom:
            update_fields.append('geom')
        if properties and properties != instance.properties:
            update_fields.append('properties')
        return update_fields

    def update(self, instance, validated_data):
        route_description = validated_data.pop('routing_information', {})

        models.RoutingInformations.objects.update_or_create(feature=instance,
                                                            defaults={'route_description': route_description.get(
                                                                'route_description',
                                                                {})})
        update_fields = self.get_update_fields(instance, validated_data)
        for key in validated_data:
            setattr(instance, key, validated_data[key])
        instance.save(update_fields=update_fields)
        return instance

    def get_relations(self, obj):
        return [{"label": relation.name,
                 "order": relation.destination.crud_view.order,
                 "url": reverse('feature-relation',
                                args=(obj.layer_id, obj.identifier, relation.pk)),
                 'id_layer_vt': f'relation-{slugify(obj.layer.name)}-{slugify(relation.name)}',
                 "crud_view_pk": relation.destination.crud_view.pk,
                 "empty": not relation.related_features.filter(origin=obj).exists()
                 } for relation in obj.layer.relations_as_origin.all() if hasattr(relation.destination, 'crud_view')]

    def get_pictures(self, obj):
        """ Return feature linked pictures grouped by category, with urls to create / replace / delete """
        return [{
            "category": {
                "id": category.pk,
                "name": category.name,
            },
            "pictogram": category.pictogram.url if category.pictogram else None,
            "pictures": FeaturePictureSerializer(obj.pictures.filter(category=category),
                                                 many=True).data,
            "action_url": reverse('picture-list', args=(obj.identifier, ))
        } for category in models.AttachmentCategory.objects.all()]

    def get_attachments(self, obj):
        """ Return feature linked pictures grouped by category, with urls to create / replace / delete """
        return [{
            "category": {
                "id": category.pk,
                "name": category.name,
            },
            "pictogram": category.pictogram.url if category.pictogram else None,
            "attachments": FeatureAttachmentSerializer(obj.attachments.filter(category=category),
                                                       many=True).data,
            "action_url": reverse('attachment-list', args=(obj.identifier, ))
        } for category in models.AttachmentCategory.objects.all()]

    def get_title(self, obj):
        """ Get Feature title, as feature_title_property content or identifier by default """
        return obj.layer.crud_view.get_feature_title(obj)

    def get_properties(self, obj):
        """ Feature properties as form initial data format (name / value) """
        results = {}
        crud_view = obj.layer.crud_view
        groups = crud_view.feature_display_groups.all()
        original_properties = deepcopy(obj.properties)

        # get ordered groups filled
        for group in groups:
            results[group.slug] = {}
            for prop in list(group.group_properties.all().values_list('key', flat=True)):
                results[group.slug][prop] = original_properties.pop(prop, None)

        return {**results, **original_properties}

    def get_display_properties(self, obj):
        """ Feature properties to display (key / value, display value and info) """
        results = {}
        crud_view = obj.layer.crud_view
        groups = crud_view.feature_display_groups.all()

        # get ordered groups filled
        for group in groups:
            serializer = FeatureDisplayPropertyGroup(group,
                                                     context={'request': self.context.get('request'),
                                                              'feature': obj})
            results[group.slug] = serializer.data

        # add default other properties
        remained_properties = crud_view.properties.filter(group__isnull=True)
        if remained_properties:
            # reconstruct property key/value list based on layer schema
            final_properties = {
                prop.key: obj.properties.get(prop.key)
                for prop in remained_properties
            }
            editable = {prop.key: prop.editable for prop in remained_properties}
            properties = serialize_group_properties(obj, final_properties, editable)

            results['__default__'] = {
                "title": "",
                "pictogram": None,
                "order": 9999,
                "properties": properties
            }

        return results

    def get_documents(self, obj):
        serializer = DocumentFeatureSerializer(obj.layer.crud_view.templates.all(),
                                               many=True,
                                               context={'request': self.context.get('request'),
                                                        'feature': obj})
        return serializer.data

    def get_geometries(self, obj):
        """ Describe geometries and action endpoint to frontend. """
        result = {
            obj.layer.name: {
                "geom": json.loads(obj.geom.geojson),
                "geom_type": obj.layer.geom_type,
                "url": reverse('feature-detail', args=(obj.layer_id, obj.identifier)),
                "identifier": obj.identifier,
                "title": _("Main geometry")
            }
        }
        for extra_geom in obj.layer.extra_geometries.all():
            geometries = obj.extra_geometries.filter(layer_extra_geom=extra_geom)
            geometry = geometries.first()
            result[extra_geom.name] = {
                "geom": json.loads(geometry.geom.geojson),
                "geom_type": extra_geom.geom_type,
                "url": reverse('feature-detail-extra-geometry', args=(obj.layer_id, obj.identifier, geometry.pk)),
                "identifier": geometry.identifier,
                "title": extra_geom.title
            } if geometry else {
                "geom": None,
                "geom_type": extra_geom.geom_type,
                "url": reverse('feature-create-extra-geometry', args=(obj.layer_id, obj.identifier, extra_geom.pk)),
                "identifier": None,
                "title": extra_geom.title
            }
        return result

    def validate_properties(self, data):
        new_data = deepcopy(data)
        # ungroup properties
        for key, value in new_data.items():
            if isinstance(value, dict):
                # pop and explode dict
                parsed_data = data.pop(key)
                for parsed_key, parsed_value in parsed_data.items():
                    data[parsed_key] = parsed_value
        # keep parent schema validation
        super().validate_properties(data)
        return data

    def save(self, **kwargs):
        old_properties = {}
        if self.instance and self.instance.pk:
            old_properties = self.instance.properties
        super().save(**kwargs)
        # save base64 file content to storage
        store_feature_files(self.instance, old_properties)

    def to_representation(self, instance):
        data = super().to_representation(instance)
        # expose properties in groups
        data['properties'] = self.get_properties(instance)
        return data

    class Meta(FeatureSerializer.Meta):
        exclude = ('source', 'target', 'layer')
        fields = None


class CrudFeatureExtraGeomSerializer(FeatureExtraGeomSerializer):
    """ Used to create or edit extra geometry. Should return Feature detail serializer """

    def to_representation(self, instance):
        # use default CrudFeatureDetailSerializer to representation
        serializer = CrudFeatureDetailSerializer(instance.feature)
        return serializer.to_representation(instance.feature)

    class Meta(FeatureExtraGeomSerializer.Meta):
        fields = ('geom', )