FastApi

FastApi Example¶

This example shows how to use sqlalchemy-filterset with FastAPI and SQLAlchemy.

The source code is available on the Github.

Application structure¶

Application has next structure:

./
├── webapp/
│   ├── __init__.py
│   ├── application.py
│   ├── endpoints.py
│   ├── schemas.py
│   ├── models.py
│   ├── filters.py
│   ├── queries.py
│   └── tests.py
├── config.yml
├── docker-compose.yml
├── Dockerfile
└── requirements.txt

Create FastAPI application¶

webapp/application.py

from typing import Generator

from fastapi import FastAPI
from myapp.models import Base
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker

engine = create_engine("postgresql://user:password@host/database")
Base.metadata.create_all(bind=engine)
SessionLocal = sessionmaker(bind=engine)


def get_db() -> Generator[Session, None, None]:
    try:
        db = SessionLocal()
        yield db
    finally:
        db.close()


app = FastAPI()

Models¶

Here is an example of creating models for e-commerce application. More information about the models can be found in the official sqlalchemy documentation

webapp/models.py

import enum
import uuid

from sqlalchemy import Boolean, Column, Enum, ForeignKey, Numeric, String
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import declarative_base, relationship

Base = declarative_base()


class CategoryType(enum.Enum):
    foo = "foo"
    bar = "bar"


class Category(Base):
    __tablename__ = "categories"
    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    type = Column(Enum(CategoryType), nullable=False)
    title = Column(String)


class Tag(Base):
    __tablename__ = "tags"
    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    title = Column(String)


class TagToProduct(Base):
    __tablename__ = "tag_to_product"
    left_id = Column(
        UUID,
        ForeignKey("tag.id", ondelete="CASCADE"),
        primary_key=True,
    )
    right_id = Column(
        UUID,
        ForeignKey("product.id", ondelete="CASCADE"),
        primary_key=True,
    )


class Product(Base):
    __tablename__ = "products"
    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    name = Column(String)
    price = Column(Numeric)
    is_active = Column(Boolean)
    category_id = Column(
        UUID(as_uuid=True),
        ForeignKey("categories.id"),
        nullable=True,
    )
    category: Category = relationship("Category", backref="products")

Schemas¶

Add simple schemas for Product model.

webapp/schemas.py

import uuid

from pydantic import BaseModel


class ProductOut(BaseModel):
    id: uuid.UUID
    name: str
    price: float
    is_active: bool

    class Config:
        orm_mode = True

Filters¶

Add FilterSet and FilterSchema for the Product model.

webapp/filters.py

import uuid

from pydantic import BaseModel
from webapp.models import Category, CategoryType, Product, Tag, TagToProduct

from sqlalchemy_filterset import (
    Filter,
    FilterSet,
    InFilter,
    JoinStrategy,
    LimitOffsetFilter,
    MultiJoinStrategy,
    OrderingField,
    OrderingFilter,
    RangeFilter,
    SearchFilter,
)


class ProductFilterSet(FilterSet):
    id = Filter(Product.id)
    ids = InFilter(Product.id)
    name = SearchFilter(Product.name)
    price = RangeFilter(Product.price)
    is_active = Filter(Product.is_active)
    category_type = Filter(
        Category.type,
        strategy=JoinStrategy(
            Category,
            Product.category_id == Category.id,
        ),
    )
    tag_title = Filter(
        Tag.title,
        strategy=MultiJoinStrategy(
            JoinStrategy(TagToProduct, onclause=Product.id == TagToProduct.right_id),
            JoinStrategy(Tag, onclause=Tag.id == TagToProduct.left_id),
        ),
    )
    ordering = OrderingFilter(
        name=OrderingField(Product.name),
        price=OrderingField(Product.price),
    )
    limit_offset = LimitOffsetFilter()


class ProductFilterSchema(BaseModel):
    id: uuid.UUID | None
    ids: list[uuid.UUID] | None
    name: str | None
    price: tuple[float, float] | None
    is_active: bool | None
    category_type: CategoryType | None
    tag_title: str | None
    ordering: list[str] | None
    limit_offset: tuple[int | None, int | None] | None

    class Config:
        orm_mode = True

Queries¶

Add queries for api endpoints.

webapp/queries.py

import dataclasses
import uuid

from fastapi import Query
from myapp.models import CategoryType


@dataclasses.dataclass
class ProductQuery:
    id: uuid.UUID | None = Query(None)
    ids: list[uuid.UUID] | None = Query(None)
    name: str | None = Query(None)
    price_min: int | None = Query(None)
    price_max: int | None = Query(None)
    is_active: bool | None = Query(None)
    category_type: CategoryType | None = Query(None)
    ordering: list[str] | None = Query(None)
    limit: int | None = Query(None)
    offset: int | None = Query(None)

    @property
    def limit_offset(self) -> tuple[int | None, int | None] | None:
        if self.limit or self.offset:
            return self.limit, self.offset
        return None

    @property
    def price(self) -> tuple[float, float] | None:
        if self.price_min and self.price_max:
            return self.price_min, self.price_max
        return None

Endpoints¶

Add endpoints for the application.

webapp/endpoints.py

from fastapi import Depends
from pydantic import parse_obj_as
from sqlalchemy import select
from sqlalchemy.orm import Session
from webapp.applications import app, get_db
from webapp.filters import ProductFilterSchema, ProductFilterSet
from webapp.models import Product
from webapp.queries import ProductQuery
from webapp.schemas import ProductOut


@app.get("/products/")
def list_products(
    filters: ProductQuery = Depends(),
    db: Session = Depends(get_db),
) -> list[ProductOut]:
    filter_set = ProductFilterSet(db, select(Product))
    filter_params = parse_obj_as(ProductFilterSchema, filters)
    filtered_products = filter_set.filter(filter_params.dict(exclude_none=True))
    return parse_obj_as(list[ProductOut], filtered_products)

Usage¶

Run application with uvicorn.

uvicorn webapp.application:app --reload

Go to http://127.0.0.1:8000/docs

FastApi docs

Conclusion¶

In conclusion, this example demonstrates how to use the sqlalchemy-filterset library with FastAPI and SQLAlchemy. It shows the structure of an application that uses this library, including the creation of a FastAPI application, models, schemas, and filters. The example also includes the usage of various filters such as InFilter, RangeFilter, and OrderingFilter to filter data from the database. The example also demonstrates the use of JoinStrategy to filter data from related models.