pythonpostgresqlsqlalchemyfastapipydantic

Problem with multiple Query in view of FastAPI application endpoint


I'm trying to develop filtering/ordering/pagination functionality for FastAPI applications. For now I'm facing difficulty with separating filtering and sorting. The code below generates undesirable swagger:

from fastapi import FastAPI, Query
from pydantic import BaseModel
from sqlalchemy import Select, create_engine, MetaData, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

app = FastAPI()

url = "postgresql+psycopg2://postgres:postgres@localhost:5432/database"
engine = create_engine(url, echo=True)


class Base(DeclarativeBase):
    metadata = MetaData()


class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    first_name: Mapped[str]
    last_name: Mapped[str]


class FilterSet(BaseModel):
    first_name: str | None

    def filter_queryset(self, query: Select) -> Select:
        conditions = self.model_dump(exclude_unset=True)
        # apply ordering
        return query


class Ordering(BaseModel):
    ordering: list[str] | None = None

    def order_queryset(self, query: Select) -> Select:
        self.ordering
        # apply ordering
        return query


@app.get("/")
async def index(filterset: FilterSet = Query(), ordering: Ordering = Query()):
    query = select(User)
    query = filterset.filter_queryset(query)
    query = ordering.order_queryset(query)
    # request database

The bad swagger:

enter image description here

Is it possible to fix it without combining FilterSet and Ordering classes into a single class?


Solution

  • One away to achieve what you want is to shift the Query annotations into the models and use them via Depends. Here's a stripped-down example (since sqlalchemy isn't really related to your problem):

    from typing import Annotated
    from fastapi import Depends, FastAPI, Query
    from pydantic import BaseModel, Field
    
    app = FastAPI()
    
    class FilterSet(BaseModel):
        first_name: str | None = Field(Query())
    
    class Ordering(BaseModel):
        ordering: list[str] = Field(Query(None))
    
    @app.get("/")
    async def index(
        filterset: Annotated[FilterSet, Depends()],
        ordering: Annotated[Ordering, Depends()],
    ):
        return [filterset.first_name, ordering.ordering]
    

    If you also want to be able to instantiate these classes yourself, you probably don't want the Query() objects hanging around as the defaults; so consider something like

    class Ordering(BaseModel):
        ordering: list[str] | None = None
    
        @staticmethod
        def from_query(ordering: list[str] = Query(None)):
            return Ordering(ordering=ordering)
    
    
    @app.get("/")
    async def index(
        filterset: Annotated[FilterSet, Depends(FilterSet.from_query)],
        ordering: Annotated[Ordering, Depends(Ordering.from_query)],
    ):
        return [filterset.first_name, ordering.ordering]