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 | def register_oapif_viewset(
key: Optional[str] = None,
serialize_geom_in_db: Optional[bool] = True,
geom_field: [str] = "geom",
crs: Optional[int] = None,
custom_serializer_attrs: Dict[str, Any] = None,
custom_viewset_attrs: Dict[str, Any] = None,
) -> Callable[[Any], models.Model]:
"""
This decorator takes care of all boilerplate code (creating a serializer, a viewset and registering it) to register
a model to the default OAPIF endpoint.
- key: allows to pass a custom name for the collection (defaults to the model's label)
- serialize_geom_in_db: delegate the geometry serialization to the DB
- geom_field: the geometry field name. If None, a null geometry is produced
- crs: the EPSG code, if empty CRS84 is assumed
- custom_serializer_attrs: allows to pass custom attributes to set to the serializer's Meta (e.g. custom fields)
- custom_viewset_attrs: allows to pass custom attributes to set to the viewset (e.g. custom pagination class)
"""
if custom_serializer_attrs is None:
custom_serializer_attrs = {}
if custom_viewset_attrs is None:
custom_viewset_attrs = {}
def inner(Model):
"""
Create the serializers
"""
Model.crs = crs
if serialize_geom_in_db and geom_field:
class AutoSerializer(GeoFeatureModelSerializer):
_geom_geojson = serializers.JSONField(required=False, allow_null=True, read_only=True)
class Meta:
model = Model
exclude = [geom_field]
geo_field = "_geom_geojson"
def to_internal_value(self, data):
# TODO: this needs improvement!!!
geo = None
if "geometry" in data:
geo = data["geometry"]
if crs not in geo:
geo["crs"] = {"type": "name", "properties": {"name": f"urn:ogc:def:crs:EPSG::{Model.crs}"}}
data = super().to_internal_value(data)
if geo:
data[geom_field] = GEOSGeometry(json.dumps(geo))
return data
else:
class AutoSerializer(GeoFeatureModelSerializer):
class Meta:
model = Model
fields = "__all__"
geo_field = geom_field
def to_internal_value(self, data):
# TODO: this needs improvement!!!
if "geometry" in data and "crs" not in data["geometry"]:
data["geometry"]["crs"] = {
"type": "name",
"properties": {"name": f"urn:ogc:def:crs:EPSG::{Model.crs}"},
}
data = super().to_internal_value(data)
return data
# Create the viewset
class OgcAPIFeatureViewSet(OAPIFDescribeModelViewSetMixin, viewsets.ModelViewSet):
queryset = Model.objects.all()
serializer_class = AutoSerializer
# TODO: these should probably be moved to the mixin
oapif_title = Model._meta.verbose_name
oapif_description = Model.__doc__
oapif_geom_lookup = geom_field
filter_backends = [BboxFilterBackend]
# Allowing '.' and '-' in urls
lookup_value_regex = r"[\w.-]+"
# Metadata
metadata_class = OAPIFMetadata
def get_success_headers(self, data):
location = reverse.reverse(f"{self.basename}-detail", {"lookup": data[Model._meta.pk.column]})
headers = {"Location": location}
return headers
def finalize_response(self, request, response, *args, **kwargs):
response = super().finalize_response(request, response, *args, **kwargs)
if request.method == "OPTIONS":
allowed_actions = self.metadata_class().determine_actions(request, self)
allowed_actions = ", ".join(allowed_actions.keys())
response.headers["Allow"] = allowed_actions
return response
def get_queryset(self):
qs = super().get_queryset()
if serialize_geom_in_db and geom_field:
qs = qs.annotate(_geom_geojson=Cast(AsGeoJSON(geom_field, False, False), models.JSONField()))
return qs
# Apply custom serializer attributes
for k, v in custom_serializer_attrs.items():
setattr(AutoSerializer.Meta, k, v)
# Apply custom viewset attributes
for k, v in custom_viewset_attrs.items():
setattr(OgcAPIFeatureViewSet, k, v)
# Register the model
oapif_router.register(key or Model._meta.label_lower, OgcAPIFeatureViewSet, key or Model._meta.label_lower)
return Model
return inner
|