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

Question

问题

我正在尝试使用FastAPI框架创建一个API，该框架允许从PostgreSQL/PostGIS数据库获取数据。我遵循了FastAPI SQLtutorial，直到现在，我一直在创建基本结构

然而，我现在面临着一个我没有设法解决的问题：我想通过一条路线提供关于一个对象的基本信息，包括它的地理信息（目前的几何体和面积）

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

我的第一次尝试是将函数添加到crud文件中的查询中。但它不起作用。查询函数似乎返回了pydantic验证器无法识别的两个对象。我试图在internet上找到其他可能性，但提供的解决方案（hybrid_属性、column_属性）主要用于简单的数据类型，允许在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()

积垢.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)

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