CODEWITHSUNDEEP
pythonsqlalchemygeojsonfastapipydantic
dlqmap
dlqmap

Reputation: 101

fastapi: mapping sqlalchemy database model to pydantic geojson feature

I just started playing with FastAPI, SQLAlchemy, Pydantic and I'm trying to build a simple API endpoint to return the rows in a postgis table as a geojson feature collection.

This is my sqlalchemy model:

class Poi(Base):
    __tablename__ = 'poi'

    id = Column(Integer, primary_key=True)
    name = Column(Text, nullable=False)
    type_id = Column(Integer)
    geometry = Column(Geometry('POINT', 4326, from_text='ST_GeomFromEWKT'),
                      nullable=False)

Using geojson_pydantic the relevant pydantic models are:

from geojson_pydantic.features import Feature, FeatureCollection
from geojson_pydantic.geometries import Point
from typing import List

class PoiProperties(BaseModel):
    name: str
    type_id: int

class PoiFeature(Feature):
    id: int
    geometry: Point
    properties: PoiProperties

class PoiCollection(FeatureCollection):
    features: List[PoiFeature]

Desired Output:

Ideally I'd like to be able to retrieve and return the database records like so:

def get_pois(db: Session, skip: int = 0, limit: int = 100):
    return db.query(Poi).offset(skip).limit(limit).all()


@app.get("/geojson", response_model=PoiCollection)
def read_geojson(skip: int = 0,
                 limit: int = 100,
                 db: Session = Depends(get_db)):
    return get_pois(db, skip=skip, limit=limit)

Still I'm trying to figure out how to map the name and type_id columns from the db model to the PoiProperties in the PoiFeature object.

Upvotes: 10

Views: 3343

Answers (1)

fchancel
fchancel

Reputation: 2699

You want to return the PoiCollection schema (response_model=schemas.PoiCollection) except that you return your database response directly without any formatting. So you have to convert your crud response into your schema response.

# Different function for translate db response to Pydantic response according to your different schema
def make_response_poi_properties(poi):
    return PoiFeature(name=poi.name, type_id=poi.type_id) 

def make_response_poi_feature(poi):
    return PoiFeature(id=poi.id, geometry=poi.geometry,properties=make_response_poi_properties(poi)) 

def make_response_poi_collection(pois):
    response = []
    for poi in pois:
        response.append(make_response_poi_feature(poi)
    return response

@app.get("/geojson", response_model=PoiCollection)
def read_geojson(skip: int = 0,
                 limit: int = 100,
                 db: Session = Depends(get_db)):
    
    # Call function for translate db data to pydantic data according to your response_model
    return make_response_poi_collection(get_pois(db, skip=skip, limit=limit))

or simply use the orm mode inside your different schema class

Upvotes: 4

Related Questions