在Flask Marshmallow中实现WKB到WKT/GeoJSON的序列化

我懂你现在卡在哪了——用Flask-SQLAlchemy建了带WKB类型geom字段的GIS模型，想通过Flask-Marshmallow把这个字段转成WKT或GeoJSON给API用，但之前的尝试没起效对吧？下面给你两种靠谱的解决方案，分别对应WKT和GeoJSON格式：

方案1：序列化为WKT格式 #

你需要自定义一个Marshmallow字段，专门处理GeoAlchemy2的WKBElement类型，直接提取它的WKT表示就行：

from marshmallow import fields
from geoalchemy2.elements import WKBElement

class WKTField(fields.Field):
    def _serialize(self, value, attr, obj, **kwargs):
        if value is None:
            return None
        # WKBElement自带wkt属性，直接返回就搞定
        return value.wkt

接着在你的Schema里替换默认的geom字段：

from marshmallow_sqlalchemy import SQLAlchemyAutoSchema
from your_app.models import YourGISModel  # 换成你的实际模型类

class YourGISModelSchema(SQLAlchemyAutoSchema):
    # 用自定义字段覆盖自动生成的geom字段
    geom = WKTField()

    class Meta:
        model = YourGISModel
        include_fk = True  # 有外键关联的话打开这个
        load_instance = True

这样序列化模型实例时，geom字段就会输出标准的WKT字符串了。

方案2：序列化为GeoJSON格式 #

如果需要输出GeoJSON，有两种方式可选，看你更偏向哪一种：

方法A：利用__geo_interface__属性 #

GeoAlchemy2的WKBElement自带符合GeoJSON规范的__geo_interface__属性，直接用它就行：

from marshmallow import fields
from geoalchemy2.elements import WKBElement

class GeoJSONField(fields.Field):
    def _serialize(self, value, attr, obj, **kwargs):
        if value is None:
            return None
        # 返回标准的GeoJSON Feature结构字典
        return value.__geo_interface__

然后在Schema里把geom字段换成GeoJSONField()，序列化后就能得到完整的GeoJSON结构。

方法B：数据库层面用ST_AsGeoJSON转换 #

如果你想在查询阶段就完成转换，可以用GeoAlchemy2的ST_AsGeoJSON函数，再在Schema里处理结果：

from geoalchemy2 import functions
from your_app import db

# 查询时直接把geom转成GeoJSON字符串
results = db.session.query(
    YourGISModel.id,
    functions.ST_AsGeoJSON(YourGISModel.geom).label("geom")
).all()

# Schema里可以直接用String字段接收，也可以转成字典
class YourGISModelSchema(SQLAlchemyAutoSchema):
    geom = fields.String()

    # 可选：把字符串转成GeoJSON字典
    @post_load
    def parse_geojson(self, data, **kwargs):
        if data.get("geom"):
            data["geom"] = json.loads(data["geom"])
        return data

    class Meta:
        model = YourGISModel
        include_fk = True

踩坑排查小提示 #

• 先确认你的模型里geom字段是用geoalchemy2.Geometry定义的，比如：

  from geoalchemy2 import Geometry
  from your_app import db
  
  class YourGISModel(db.Model):
      id = db.Column(db.Integer, primary_key=True)
      geom = db.Column(Geometry(geometry_type="POINT", srid=4326))
  

• 如果之前自定义字段没生效，检查是不是Schema里自动生成的geom字段覆盖了你的自定义字段——显式声明geom字段就能解决这个问题。

内容的提问来源于stack exchange，提问作者ISONecroMAn