Problem
I am trying to create an API using the FastAPI framework that allow to get data from a PostgreSQL / PostGIS database. I have followed the FastAPI SQL tutorial and have managed until now to create the basic structure.
However, I am now facing a problem that I didn't manage to solve : I would like to provide through a route the basic info about an object, including its geographic ones (geometry and area for now).
I managed to get the geometry with GeoAlchemy and parse it to pydantic validators with a custom function. But I didn't find any way to parse the area, that must be calculated through postgresql query as it is tied to a specific PostGIS function.
My first try was to add the function to the query in the crud file. But it doesn't work. It seems like the query function return two objects that the pydantic validator does not recognize. I've tried to find other possibilities on internet, but the solutions provided (hybrid_property, column_property) works mainly with simple data type that allows to create a "calculated" attribute within the python code.
Have you idea how I could bypass this problem ? Thank you for your help
Here is the code of my initial try
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)
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
And the error I get
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)