SQLAlchemy¶
fastapi-filters provides first-class SQLAlchemy integration for applying filters and sorting to
select() statements.
Installation¶
pip install fastapi-filters[sqlalchemy]
Basic Usage¶
from fastapi import Depends, FastAPI
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from fastapi_filters import FilterField, FilterSet, SortingValues, create_sorting
from fastapi_filters.ext.sqlalchemy import apply_filters, apply_sorting
app = FastAPI()
class UserFilters(FilterSet):
name: FilterField[str]
age: FilterField[int]
is_active: FilterField[bool]
@app.get("/users")
async def get_users(
db: AsyncSession = Depends(get_db),
filters: UserFilters = Depends(),
sorting: SortingValues = Depends(create_sorting("age", "created_at")),
):
stmt = select(User)
stmt = apply_filters(stmt, filters)
stmt = apply_sorting(stmt, sorting)
return (await db.scalars(stmt)).all()
Or use the combined helper:
from fastapi_filters.ext.sqlalchemy import apply_filters_and_sorting
@app.get("/users")
async def get_users(
db: AsyncSession = Depends(get_db),
filters: UserFilters = Depends(),
sorting: SortingValues = Depends(create_sorting("age", "created_at")),
):
stmt = apply_filters_and_sorting(select(User), filters, sorting)
return (await db.scalars(stmt)).all()
Additional Namespace¶
Use additional to add computed or derived columns that filters/sorting can reference:
from sqlalchemy import func
@app.get("/users")
async def get_users(
filters: UserFilters = Depends(),
sorting: SortingValues = Depends(create_sorting("age", "created_at")),
):
stmt = select(User)
stmt = apply_filters_and_sorting(
stmt,
filters,
sorting,
additional={
"name": func.lower(User.name), # filter/sort on lowercased name
},
)
...
You can also use FilterField instances as keys:
stmt = apply_filters_and_sorting(
stmt,
filters,
sorting,
additional={
UserFilters.name: func.lower(User.name),
},
)
Field Remapping¶
Use remapping when the filter field name differs from the database column name:
@app.get("/users")
async def get_users(filters: UserFilters = Depends()):
stmt = apply_filters(
select(User),
filters,
remapping={"name": "full_name"}, # filter "name" maps to column "full_name"
)
...
Custom Filter Logic¶
Per-Call Custom Filters¶
Pass apply_filter to handle specific field/operator combinations:
def my_apply_filter(stmt, ns, field, op, val):
if field == "search" and op == FilterOperator.eq:
return stmt.where(
User.name.ilike(f"%{val}%") | User.email.ilike(f"%{val}%")
)
raise NotImplementedError # fall back to default
stmt = apply_filters(
select(User),
filters,
apply_filter=my_apply_filter,
)
Per-Call Custom Condition¶
Pass add_condition to control how conditions are added to the statement:
def my_add_condition(stmt, field, condition):
if field == "tags":
return stmt.where(condition) # custom WHERE logic
raise NotImplementedError # fall back to default
stmt = apply_filters(
select(User),
filters,
add_condition=my_add_condition,
)
Generate Filters from ORM Model¶
Auto-generate filters directly from a SQLAlchemy ORM model:
from fastapi import Depends
from fastapi_filters.ext.sqlalchemy import create_filters_from_orm
from fastapi_filters import FilterValues
@app.get("/users")
async def get_users(
filters: FilterValues = Depends(create_filters_from_orm(User)),
):
stmt = apply_filters(select(User), filters)
...
Options:
create_filters_from_orm(
User,
include={"name", "age"}, # only these fields
exclude={"password_hash"}, # exclude these fields
include_fk=True, # include foreign key columns
remapping={"name": "user_name"}, # rename fields
)
Tip
Prefer using FilterSet over create_filters_from_orm for new code. FilterSet gives you
type-safe field access, operator overloading, subset()/extract() methods, and inheritance.
Generate Sorting from ORM Model¶
from fastapi_filters.ext.sqlalchemy import create_sorting_from_orm
@app.get("/users")
async def get_users(
sorting: SortingValues = Depends(create_sorting_from_orm(User, default="+age")),
):
...
Full Example¶
from datetime import datetime
from typing import Any
from fastapi import Depends, FastAPI
from sqlalchemy import Integer, func, select
from sqlalchemy.dialects.postgresql import ARRAY
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, MappedAsDataclass, mapped_column
from fastapi_filters import FilterField, FilterSet, SortingValues, create_sorting
from fastapi_filters.ext.sqlalchemy import apply_filters_and_sorting
class Base(MappedAsDataclass, DeclarativeBase):
pass
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column()
age: Mapped[int] = mapped_column()
marks: Mapped[list[int]] = mapped_column(ARRAY(Integer))
is_active: Mapped[bool] = mapped_column()
created_at: Mapped[datetime] = mapped_column()
class UserFilters(FilterSet):
name: FilterField[str]
age: FilterField[int]
marks: FilterField[list[int]]
is_active: FilterField[bool]
created_at: FilterField[datetime]
app = FastAPI()
@app.get("/users")
async def get_users(
db: AsyncSession = Depends(get_db),
filters: UserFilters = Depends(),
sorting: SortingValues = Depends(create_sorting("age", "created_at")),
) -> Any:
stmt = apply_filters_and_sorting(
select(User),
filters,
sorting,
additional={
"name": func.lower(User.name),
},
)
return (await db.scalars(stmt)).all()