Geoalchemy2 pydantic 几何架构（FastAPI）

Question

我想将 PostGIS 与 FastAPI 结合使用，因此使用 geoalchemy2 与 alembic 在数据库中创建表。 但我无法正确声明 pydantic v2 中的架构。

我的代码如下所示：

# auto-generated from env.py
from alembic import op
import sqlalchemy as sa
from geoalchemy2 import Geometry
from sqlalchemy.dialects import postgresql

...

def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_geospatial_table('solarpark',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('name_of_model', sa.String(), nullable=True),
    sa.Column('comment', sa.String(), nullable=True),
    sa.Column('lat', sa.ARRAY(sa.Float()), nullable=True),
    sa.Column('lon', sa.ARRAY(sa.Float()), nullable=True),
    sa.Column('geom', Geometry(geometry_type='POLYGON', srid=4326, spatial_index=False, from_text='ST_GeomFromEWKT', name='geometry'), nullable=True),
    sa.PrimaryKeyConstraint('id')
    )
    op.create_geospatial_index('idx_solarpark_geom', 'solarpark', ['geom'], unique=False, postgresql_using='gist', postgresql_ops={})
    op.create_index(op.f('ix_solarpark_id'), 'solarpark', ['id'], unique=False)
    # ### end Alembic commands ###

# models.py

from geoalchemy2 import Geometry
from sqlalchemy import ARRAY, Column, Date, Float, Integer, String

from app.db.base_class import Base


class SolarPark(Base):
    id = Column(Integer, primary_key=True, index=True)
    name_of_model = Column(String)
    comment = Column(String, default="None")
    lat = Column(ARRAY(item_type=Float))
    lon = Column(ARRAY(item_type=Float))
    geom = Column(Geometry("POLYGON", srid=4326))

# schemas.py

from typing import List

from pydantic import ConfigDict, BaseModel, Field

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: [WHAT TO INSERT HERE?] = Field('POLYGON ((599968.55 5570202.63, 599970.90 5570205.59, 599973.65 5570203.42, 599971.31 5570200.46, 599968.55 5570202.63))')

我希望列 geom 是一种几何类型，可以对其执行空间操作。但我如何在 pydantic v2 中声明这一点？

提前非常感谢！

Answer 1

所以我找到了答案：

# schemas.py
from typing import List

from pydantic import ConfigDict, BaseModel, Field

from geoalchemy2.types import WKBElement
from typing_extensions import Annotated

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: Annotated[str, WKBElement] = Field('POLYGON ((599968.55 5570202.63, 599970.90 5570205.59, 599973.65 5570203.42, 599971.31 5570200.46, 599968.55 5570202.63))')

但是你还需要改变你的CRUD函数：

# crud_solarpark.py
class CRUDSolarPark(CRUDBase[SolarPark, SolarParkCreate, SolarParkUpdate]):
    def get(self, db: Session, *, id: int) -> SolarPark:
        db_obj = db.query(SolarPark).filter(SolarPark.id == id).first()
        if db_obj is None:
            return None

        if isinstance(db_obj.geom, str):
            db_obj.geom = WKTElement(db_obj.geom)
        db_obj.geom = to_shape(db_obj.geom).wkt
        return db_obj

    def get_multi(self, db: Session, *, skip: int = 0, limit: int = 100) -> SolarPark:
        db_obj = db.query(SolarPark).offset(skip).limit(limit).all()
        if db_obj is None:
            return None

        for obj in db_obj:
            if isinstance(obj.geom, str):
                obj.geom = WKTElement(obj.geom)
            obj.geom = to_shape(obj.geom).wkt
        return db_obj

    def create(self, db: Session, *, obj_in: SolarParkCreate) -> SolarPark:
        obj_in_data = jsonable_encoder(obj_in)
        db_obj = SolarPark(**obj_in_data)  # type: ignore
        db.add(db_obj)
        db.commit()
        db.refresh(db_obj)
        db_obj.geom = to_shape(db_obj.geom).wkt
        return db_obj

    def update(
        self,
        db: Session,
        *,
        db_obj: SolarPark,
        obj_in: Union[SolarParkUpdate, Dict[str, Any]],
    ) -> SolarPark:
        obj_data = jsonable_encoder(db_obj)
        if isinstance(obj_in, dict):
            update_data = obj_in
        else:
            update_data = obj_in.dict(exclude_unset=True)
        for field in obj_data:
            if field in update_data:
                setattr(db_obj, field, update_data[field])
        db.add(db_obj)
        db.commit()
        db.refresh(db_obj)
        db_obj.geom = to_shape(db_obj.geom).wkt
        return db_obj

希望对使用 FastAPI 和 PostGIS 的任何人有所帮助:)