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

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

问题

我正在尝试使用 FastAPI 框架创建一个 API，该框架允许从 Postgresql / PostGIS 数据库中获取数据。我遵循了 FastAPI sql tutorial 并且到目前为止已经设法创建了基本结构。

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

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

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

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

这是我第一次尝试的代码

database.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()

models.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)

schemas.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)

ma​​in.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,send)
  File "/upfactor/code/api/info/venv/lib/python3.9/site-packages/starlette/applications.py",line 111,in __call__
    await self.middleware_stack(scope,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,_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,sender)
  File "/upfactor/code/api/info/venv/lib/python3.9/site-packages/starlette/routing.py",line 566,in __call__
    await route.handle(scope,send)
  File "/upfactor/code/api/info/venv/lib/python3.9/site-packages/starlette/routing.py",line 227,in handle
    await self.app(scope,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)