1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231
|
from collections import OrderedDict
from django.contrib.gis.geos import Polygon
from django.core.exceptions import ImproperlyConfigured
from rest_framework.serializers import (
LIST_SERIALIZER_KWARGS,
ListSerializer,
ModelSerializer,
)
from .fields import GeometryField, GeometrySerializerMethodField # noqa
class GeoModelSerializer(ModelSerializer):
"""
Deprecated, will be removed in django-rest-framework-gis 1.0
"""
class GeoFeatureModelListSerializer(ListSerializer):
@property
def data(self):
return super(ListSerializer, self).data
def to_representation(self, data):
"""
Add GeoJSON compatible formatting to a serialized queryset list
"""
return OrderedDict(
(
("type", "FeatureCollection"),
("features", super().to_representation(data)),
)
)
class GeoFeatureModelSerializer(ModelSerializer):
"""
A subclass of ModelSerializer
that outputs geojson-ready data as
features and feature collections
"""
@classmethod
def many_init(cls, *args, **kwargs):
child_serializer = cls(*args, **kwargs)
list_kwargs = {'child': child_serializer}
list_kwargs.update(
{
key: value
for key, value in kwargs.items()
if key in LIST_SERIALIZER_KWARGS
}
)
meta = getattr(cls, 'Meta', None)
list_serializer_class = getattr(
meta, 'list_serializer_class', GeoFeatureModelListSerializer
)
return list_serializer_class(*args, **list_kwargs)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
meta = getattr(self, 'Meta')
default_id_field = None
primary_key = self.Meta.model._meta.pk.name
# use primary key as id_field when possible
if (
not hasattr(meta, 'fields')
or meta.fields == '__all__'
or primary_key in meta.fields
):
default_id_field = primary_key
meta.id_field = getattr(meta, 'id_field', default_id_field)
if not hasattr(meta, 'geo_field'):
raise ImproperlyConfigured(
"You must define a 'geo_field'. "
"Set it to None if there is no geometry."
)
def check_excludes(field_name, field_role):
"""make sure the field is not excluded"""
if hasattr(meta, 'exclude') and field_name in meta.exclude:
raise ImproperlyConfigured(
"You cannot exclude your '{0}'.".format(field_role)
)
def add_to_fields(field_name):
"""Make sure the field is included in the fields"""
if hasattr(meta, 'fields') and meta.fields != '__all__':
if field_name not in meta.fields:
if type(meta.fields) is tuple:
additional_fields = (field_name,)
else:
additional_fields = [field_name]
meta.fields += additional_fields
check_excludes(meta.geo_field, 'geo_field')
if meta.geo_field is not None:
add_to_fields(meta.geo_field)
meta.bbox_geo_field = getattr(meta, 'bbox_geo_field', None)
if meta.bbox_geo_field:
check_excludes(meta.bbox_geo_field, 'bbox_geo_field')
add_to_fields(meta.bbox_geo_field)
meta.auto_bbox = getattr(meta, 'auto_bbox', False)
if meta.bbox_geo_field and meta.auto_bbox:
raise ImproperlyConfigured(
"You must eiher define a 'bbox_geo_field' or "
"'auto_bbox', but you can not set both"
)
def to_representation(self, instance):
"""
Serialize objects -> primitives.
"""
# prepare OrderedDict geojson structure
feature = OrderedDict()
# keep track of the fields being processed
processed_fields = set()
# optional id attribute
if self.Meta.id_field:
field = self.fields[self.Meta.id_field]
value = field.get_attribute(instance)
feature["id"] = field.to_representation(value)
processed_fields.add(self.Meta.id_field)
# required type attribute
# must be "Feature" according to GeoJSON spec
feature["type"] = "Feature"
# geometry attribute
# must be present in output according to GeoJSON spec
if self.Meta.geo_field:
field = self.fields[self.Meta.geo_field]
geo_value = field.get_attribute(instance)
feature["geometry"] = field.to_representation(geo_value)
processed_fields.add(self.Meta.geo_field)
else:
feature["geometry"] = None
# Bounding Box
# if auto_bbox feature is enabled
# bbox will be determined automatically automatically
if self.Meta.auto_bbox and geo_value:
feature["bbox"] = geo_value.extent
# otherwise it can be determined via another field
elif self.Meta.bbox_geo_field:
field = self.fields[self.Meta.bbox_geo_field]
value = field.get_attribute(instance)
feature["bbox"] = value.extent if hasattr(value, 'extent') else None
processed_fields.add(self.Meta.bbox_geo_field)
# the list of fields that will be processed by get_properties
# we will remove fields that have been already processed
# to increase performance on large numbers
fields = [
field_value
for field_key, field_value in self.fields.items()
if field_key not in processed_fields
]
# GeoJSON properties
feature["properties"] = self.get_properties(instance, fields)
return feature
def get_properties(self, instance, fields):
"""
Get the feature metadata which will be used for the GeoJSON
"properties" key.
By default it returns all serializer fields excluding those used for
the ID, the geometry and the bounding box.
:param instance: The current Django model instance
:param fields: The list of fields to process (fields already processed have been removed)
:return: OrderedDict containing the properties of the current feature
:rtype: OrderedDict
"""
properties = OrderedDict()
for field in fields:
if field.write_only:
continue
value = field.get_attribute(instance)
representation = None
if value is not None:
representation = field.to_representation(value)
properties[field.field_name] = representation
return properties
def to_internal_value(self, data):
"""
Override the parent method to first remove the GeoJSON formatting
"""
if 'properties' in data:
data = self.unformat_geojson(data)
return super().to_internal_value(data)
def unformat_geojson(self, feature):
"""
This function should return a dictionary containing keys which maps
to serializer fields.
Remember that GeoJSON contains a key "properties" which contains the
feature metadata. This should be flattened to make sure this
metadata is stored in the right serializer fields.
:param feature: The dictionary containing the feature data directly
from the GeoJSON data.
:return: A new dictionary which maps the GeoJSON values to
serializer fields
"""
attrs = feature["properties"]
if 'geometry' in feature and self.Meta.geo_field:
attrs[self.Meta.geo_field] = feature['geometry']
if self.Meta.id_field and 'id' in feature:
attrs[self.Meta.id_field] = feature['id']
if self.Meta.bbox_geo_field and 'bbox' in feature:
attrs[self.Meta.bbox_geo_field] = Polygon.from_bbox(feature['bbox'])
return attrs
|