FastAPI中如何在Pydantic v2正确声明Geoalchemy2 Geometry类型？

问题描述 #

使用FastAPI+PostGIS+Geoalchemy2+Alembic搭建项目，数据库表已通过Alembic创建完成，但无法在Pydantic v2的Schema中正确声明geom字段类型，需要该字段支持空间操作的类型转换与验证。

解决方案 #

Pydantic v2本身不原生支持Geoalchemy2的Geometry类型，需通过自定义Pydantic类型实现WKT字符串与Geoalchemy Geometry对象的双向转换，同时保证字段验证和序列化的正确性。

步骤1：创建自定义Pydantic类型 #

新建一个自定义类型文件（比如custom_types.py），实现对Geometry类型的验证和序列化逻辑：

from pydantic import GetCoreSchemaHandler
from pydantic_core import core_schema
from geoalchemy2 import WKTElement, Geometry
from typing import Any, Union

class GeometryType:
    @classmethod
    def __get_pydantic_core_schema__(cls, source_type: Any, handler: GetCoreSchemaHandler) -> core_schema.CoreSchema:
        # 验证逻辑：将输入转换为Geoalchemy Geometry对象
        def validate(value: Union[str, Geometry]) -> Geometry:
            if isinstance(value, str):
                # 把WKT字符串转为带SRID的WKTElement（匹配数据库中的SRID=4326）
                return WKTElement(value, srid=4326)
            elif isinstance(value, Geometry):
                # 已经是Geometry对象，直接返回
                return value
            else:
                raise ValueError("geom字段必须是WKT格式字符串或Geoalchemy Geometry对象")

        # 序列化逻辑：将Geometry对象转为WKT字符串供前端接收
        def serialize(value: Geometry) -> str:
            return value.wkt if hasattr(value, 'wkt') else str(value)

        return core_schema.no_info_wrap_validator_function(
            validate,
            core_schema.union_schema([
                core_schema.str_schema(),
                core_schema.is_instance_schema(Geometry)
            ]),
            serialization=core_schema.plain_serializer_function_ser_schema(serialize)
        )

步骤2：在Schema中使用自定义类型 #

修改schemas.py，将geom字段的类型指定为自定义的GeometryType：

from typing import List
from pydantic import ConfigDict, BaseModel, Field
from .custom_types import GeometryType  # 导入自定义类型

class SolarParkBase(BaseModel):
    model_config = ConfigDict(from_attributes=True, arbitrary_types_allowed=True)

    name_of_model: str = Field("test-model")
    comment: str = "None"
    lat: List[float] = Field([599968.55, 599970.90, 599973.65, 599971.31, 599968.55])
    lon: List[float] = Field([5570202.63, 5570205.59, 5570203.42, 5570200.46, 5570202.63])
    geom: GeometryType = Field('POLYGON ((599968.55 5570202.63, 599970.90 5570205.59, 599973.65 5570203.42, 599971.31 5570200.46, 599968.55 5570202.63))')

关键说明 #

• 输入验证：支持接收WKT格式字符串（前端传入）或Geoalchemy Geometry对象（数据库读取），自动转换为统一的Geometry类型。
• 序列化输出：将数据库返回的Geometry对象转为WKT字符串，方便前端解析使用。
• 空间操作支持：在SQLAlchemy查询中，可直接使用Geoalchemy2提供的空间函数（如ST_Intersects、ST_Within等）对geom字段执行空间操作，Schema的类型转换不影响数据库层面的空间计算。

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