Skip to content

Instantly share code, notes, and snippets.

@casoetan
Forked from om-henners/serializer_utils.py
Created November 21, 2019 23:37
Show Gist options
  • Save casoetan/1789c3403b9ac95d7792cf8b8446bfe5 to your computer and use it in GitHub Desktop.
Save casoetan/1789c3403b9ac95d7792cf8b8446bfe5 to your computer and use it in GitHub Desktop.
Marshmallow serializers for turning fields into GeoJSON objects based on flask-sqlalchemy and geoalchemy2
"""
Marshmallow wrappers to produce GeoJSON instead of a flat dictionary.
"""
from flask_marshmallow import Marshmallow
import geoalchemy2
from geoalchemy2.shape import from_shape, to_shape
from marshmallow import fields, pre_load, post_dump, ValidationError
import marshmallow_sqlalchemy as msqla
from shapely import geometry
ma = Marshmallow()
class GeometryField(fields.Field):
"""
Use shapely and geoalchemy2 to serialize / deserialize a point
Does make a big assumption about the data being spat back out as
JSON, but what the hey.
"""
def _serialize(self, value, attr, obj):
if value is None:
return None
return geometry.mapping(to_shape(value))
def _deserialize(self, value, attr, data):
if value is None:
return None
return from_shape(geometry.shape(value))
msqla.ModelConverter.SQLA_TYPE_MAPPING[geoalchemy2.Geography] = GeometryField
msqla.ModelConverter.SQLA_TYPE_MAPPING[geoalchemy2.Geometry] = GeometryField
msqla.ModelConverter.SQLA_TYPE_MAPPING[geoalchemy2.Raster] = fields.Raw
msqla.ModelConverter.SQLA_TYPE_MAPPING[geoalchemy2.RasterElement] = fields.Raw
class MarshmallowGeoJSON:
"""
Base class for wrappping and unwrapping GeoJSON objects.
Reading from https://marshmallow.readthedocs.io/en/3.0/extending.html it
should be possible to wrap and unwrap envelopes around the GeoJSON data
as it comes in and out of the system.
"""
__geometry_field_name__ = 'geom' # or geom, or shape, or ....
def unwrap_feature(self, data):
"""
Unwrap an individual feature object
Pull down all the properties field, and then under the geometry
field name put in the actual geometry data
"""
if data['type'] != 'Feature':
raise ValidationError('Expecting a Feature object')
flat = data['properties']
flat[self.__geometry_field_name__] = data['geometry']
return flat
@pre_load(pass_many=True)
def unwrap_envelope(self, data, many):
if 'type' not in data:
raise ValidationError('GeoJSON type could not be found')
if many and data['type'] != 'FeatureCollection':
raise ValidationError('Expecting a FeatureCollection object')
if not many:
return self.unwrap_feature(data)
return [self.unwrap_feature(feature) for feature in data['features']]
def wrap_feature(self, data):
"""
Wrap the individual feature as a GeoJSON feature object
"""
feature = {
'type': 'Feature',
'geometry': data.pop(self.__geometry_field_name__)
}
feature['properties'] = data
return feature
@post_dump(pass_many=True)
def wrap_with_envelope(self, data, many):
if not many:
return self.wrap_feature(data)
return {
'type': 'FeatureCollection',
'features': [self.wrap_feature(feature) for feature in data]
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment