如何从 SQL 查询中的计算字段创建 SqlAlchemy 类属性？

Question

问题

我正在尝试使用 FastAPI 框架创建一个 API，该框架允许从 PostgreSQL / PostGIS 数据库获取数据。我遵循了 FastAPI SQL教程，到目前为止已经成功创建了基本结构。

但是，我现在面临一个无法解决的问题：我想通过路线提供有关对象的基本信息，包括其地理信息（目前是几何形状和面积）。

我设法使用 GeoAlchemy 获取几何图形，并使用自定义函数将其解析为 pydantic 验证器。但我没有找到任何方法来解析该区域，该区域必须通过 postgresql 查询来计算，因为它与特定的 PostGIS 函数相关联。

我的第一次尝试是将函数添加到 crud 文件中的查询中。但这不起作用。查询函数似乎返回了 pydantic 验证器无法识别的两个对象。我试图在互联网上寻找其他可能性，但提供的解决方案（hybrid_property、column_property）主要适用于简单的数据类型，允许在 python 代码中创建“计算”属性。

你知道我如何绕过这个问题吗？感谢您的帮助

这是我最初尝试的代码

数据库.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

RAW_DB_CONNECTION = "**************"

engine = create_engine(RAW_DB_CONNECTION)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

crud.py

from sqlalchemy import func
from sqlalchemy.orm import Session
from . import models, schemas


def parcelle_get_basic_info(db: Session, id: str):
    return db.query(
        models.Parcelle,
        func.ST_Area(func.ST_Transform(models.Parcelle.geom, 2154)).label('area')
        ).filter(
            models.Parcelle.id == id
        ).first()

模型.py

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String, select, func
from geoalchemy2 import Geometry
from sqlalchemy.orm import query_expression
from sqlalchemy.sql import literal

from .database import Base

class Parcelle(Base):
    __tablename__= "_limites_parcelle_cadastre"
    __table_args__ = {'schema': 'admin'}

    id = Column(String, primary_key=True, index=True)
    commune = Column(String)
    geom = Column(Geometry(geometry_type='MULTIPOLYGON', srid=4326))
    area = Column(Float)

模式.py

from typing import List, Optional
from geoalchemy2.shape import to_shape 
from geoalchemy2.elements import WKBElement
from pydantic import BaseModel, validator

def ewkb_to_wkt(geom: WKBElement):
    """
    Converts a geometry formated as WKBE to WKT 
    in order to parse it into pydantic Model

    Args:
        geom (WKBElement): A geometry from GeoAlchemy query
    """
    return to_shape(geom).wkt

class ItemBase(BaseModel):
    id: str
    commune: str
    geom : str
    area : Optional[float]

    class Config:
        orm_mode = True
    
    @validator('geom', pre=True,allow_reuse=True,whole=True, always=True)
    def correct_geom_format(cls, v):
        if not isinstance(v, WKBElement):
            raise ValueError('must be a valid WKBE element')
        return ewkb_to_wkt(v)

主要.py

from typing import List

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/parcelle/{parcelle_id}", response_model=schemas.ItemBase)
def get_parcelle_info(parcelle_id: str, db: Session = Depends(get_db)):
    parcelle_info = crud.parcelle_get_basic_info(db, id=parcelle_id)
    if parcelle_info is None:
        raise HTTPException(status_code=404, detail="Parcelle not found")
    return parcelle_info

我得到的错误

127.0.0.1:59404 - "GET /parcelle/33449000CE0006 HTTP/1.1" 500 Internal Server Error
ERROR:    Exception in ASGI application
Traceback (most recent call last):
  File "/upfactor/code/api/info/venv/lib/python3.9/site-packages/uvicorn/protocols/http/h11_impl.py", line 394, in run_asgi
    result = await app(self.scope, self.receive, self.send)
  File "/upfactor/code/api/info/venv/lib/python3.9/site-packages/uvicorn/middleware/proxy_headers.py", line 45, in __call__
    return await self.app(scope, receive, send)
  File "/upfactor/code/api/info/venv/lib/python3.9/site-packages/fastapi/applications.py", line 199, in __call__
    await super().__call__(scope, receive, send)
  File "/upfactor/code/api/info/venv/lib/python3.9/site-packages/starlette/applications.py", line 111, in __call__
    await self.middleware_stack(scope, receive, send)
  File "/upfactor/code/api/info/venv/lib/python3.9/site-packages/starlette/middleware/errors.py", line 181, in __call__
    raise exc from None
  File "/upfactor/code/api/info/venv/lib/python3.9/site-packages/starlette/middleware/errors.py", line 159, in __call__
    await self.app(scope, receive, _send)
  File "/upfactor/code/api/info/venv/lib/python3.9/site-packages/starlette/exceptions.py", line 82, in __call__
    raise exc from None
  File "/upfactor/code/api/info/venv/lib/python3.9/site-packages/starlette/exceptions.py", line 71, in __call__
    await self.app(scope, receive, sender)
  File "/upfactor/code/api/info/venv/lib/python3.9/site-packages/starlette/routing.py", line 566, in __call__
    await route.handle(scope, receive, send)
  File "/upfactor/code/api/info/venv/lib/python3.9/site-packages/starlette/routing.py", line 227, in handle
    await self.app(scope, receive, send)
  File "/upfactor/code/api/info/venv/lib/python3.9/site-packages/starlette/routing.py", line 41, in app
    response = await func(request)
  File "/upfactor/code/api/info/venv/lib/python3.9/site-packages/fastapi/routing.py", line 209, in app
    response_data = await serialize_response(
  File "/upfactor/code/api/info/venv/lib/python3.9/site-packages/fastapi/routing.py", line 126, in serialize_response
    raise ValidationError(errors, field.type_)
pydantic.error_wrappers.ValidationError: 3 validation errors for ItemBase
response -> id
  field required (type=value_error.missing)
response -> commune
  field required (type=value_error.missing)
response -> geom
  field required (type=value_error.missing)