add fastapi pagination article
All checks were successful
Build Blog Docker Image / build docker (push) Successful in 2m45s

This commit is contained in:
2026-03-21 06:53:29 -04:00
parent 106df3574c
commit d2b1548dd3
9 changed files with 450 additions and 0 deletions

View File

@@ -0,0 +1,426 @@
---
title: "How to implement pagination, sorting and filtering with FastAPI ?"
date: 2026-03-21
slug: "how-to-implement-pagination-sorting-and-filtering-with-fastapi"
tags: ["python", "fastapi"]
type: "programming"
---
## Overview
Pagination, filtering, and sorting are features you end up implementing in almost every API. This ensures good performance and a modern user experience, which is what most websites use.
In this article, I'll walk through how I handle these concerns in my FastAPI projects using [fastapi-toolsets](https://github.com/d3vyce/fastapi-toolsets), a small library I built around SQLAlchemy async and Postgres.
The example we'll use is a simple article listing API with offset and cursor pagination, full-text search, facet filtering, and client-driven sorting.
## Offset vs cursor pagination
Before diving into the implementation, it's worth understanding the two pagination strategies and when to use each.
### Offset pagination
Offset pagination is the classic approach. You pass a page number and a page size, and the database skips `(page - 1) * size` rows before returning results.
**Standard REST API convention:**
```
GET /articles?page=2&items_per_page=20
```
**Typical response shape:**
```json
{
"items": [...],
"total": 143,
"page": 2,
"total_pages": 8
}
```
It's simple, predictable, and lets clients jump to any page. The downside is performance: `OFFSET` forces the database to scan and discard all preceding rows. On large tables with high offsets, this gets slow. There's also a consistency issue — if a new item is inserted while the client is paginating, pages can shift and items appear duplicated or skipped.
### Cursor pagination
Cursor pagination replaces the page number with an opaque token representing the position of the last seen item. The server uses this cursor to seek directly to the next page.
**Standard REST API convention:**
```
GET /articles?cursor=eyJpZCI6IjEyMyJ9&items_per_page=20
```
**Typical response shape:**
```json
{
"items": [...],
"next_cursor": "eyJpZCI6IjE0MyJ9",
"has_next": true
}
```
Performance stays constant regardless of how deep into the dataset you are, and the result set is stable even if rows are inserted concurrently. The trade-off: you lose the ability to jump to an arbitrary page, which makes it less suited for traditional paginated tables.
### When to use which
| | Offset | Cursor |
|---|---|---|
| Jump to arbitrary page | Yes | No |
| Performance on large datasets | Degrades with high offsets | Stable |
| Stable results under concurrent writes | No | Yes |
| Best for | Admin tables, search results | Feeds, infinite scroll, large datasets |
## Project structure
The example is organized as a standard FastAPI app:
```
pagination_search/
├── app.py # FastAPI app setup
├── db.py # SQLAlchemy async engine + session dependency
├── models.py # SQLAlchemy models
├── crud.py # CrudFactory declaration
├── schemas.py # Pydantic response schemas
└── routes.py # API routes
```
### Models
We have two models: `Category` and `Article`. The `Article` model uses the [CreatedAtMixin](https://fastapi-toolsets.d3vyce.fr/module/models/#createdatmixin) provided by `fastapi-toolsets` to automatically add a `created_at` timestamp — which we'll use later as the cursor column for cursor-based pagination.
```python
import uuid
from sqlalchemy import Boolean, ForeignKey, String, Text
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from fastapi_toolsets.models import CreatedAtMixin
class Base(DeclarativeBase):
pass
class Category(Base):
__tablename__ = "categories"
id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
name: Mapped[str] = mapped_column(String(64), unique=True)
articles: Mapped[list["Article"]] = relationship(back_populates="category")
class Article(Base, CreatedAtMixin):
__tablename__ = "articles"
id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
title: Mapped[str] = mapped_column(String(256))
body: Mapped[str] = mapped_column(Text)
status: Mapped[str] = mapped_column(String(32))
published: Mapped[bool] = mapped_column(Boolean, default=False)
category_id: Mapped[uuid.UUID | None] = mapped_column(
ForeignKey("categories.id"), nullable=True
)
category: Mapped["Category | None"] = relationship(back_populates="articles")
```
### Database setup
The database layer uses SQLAlchemy's async engine with `asyncpg`. `fastapi-toolsets` provides two helpers: [create_db_dependency](https://fastapi-toolsets.d3vyce.fr/module/db/#session-dependency) for use as a FastAPI `Depends`, and [create_db_context](https://fastapi-toolsets.d3vyce.fr/module/db/#session-context-manager) for use as an async context manager outside of request handling.
```python
from typing import Annotated
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from fastapi_toolsets.db import create_db_context, create_db_dependency
DATABASE_URL = "postgresql+asyncpg://postgres:postgres@localhost:5432/postgres"
engine = create_async_engine(url=DATABASE_URL, future=True)
async_session_maker = async_sessionmaker(bind=engine, expire_on_commit=False)
get_db = create_db_dependency(session_maker=async_session_maker)
get_db_context = create_db_context(session_maker=async_session_maker)
SessionDep = Annotated[AsyncSession, Depends(get_db)]
```
### Declaring the CRUD factory
This is where the magic happens. Instead of writing query logic in every route, [CrudFactory](https://fastapi-toolsets.d3vyce.fr/module/crud/#creating-a-crud-class) centralizes all the configuration in one place: which fields are searchable, which can be used as facet filters, and which are exposed for client-driven ordering.
```python
from fastapi_toolsets.crud import CrudFactory
from .models import Article, Category
ArticleCrud = CrudFactory(
model=Article,
cursor_column=Article.created_at, # column used for cursor pagination
searchable_fields=[ # fields included in full-text search
Article.title,
Article.body,
(Article.category, Category.name), # joined relation field
],
facet_fields=[ # fields exposed as filter dropdowns
Article.status,
(Article.category, Category.name),
],
order_fields=[ # fields the client can sort by
Article.title,
Article.created_at,
],
)
```
> The tuple syntax `(Article.category, Category.name)` tells the factory to join the `Category` table and expose `Category.name` as a searchable/filterable field — no manual join needed in routes.
### Response schema
A minimal Pydantic schema for the article list response:
```python
import datetime
import uuid
from fastapi_toolsets.schemas import PydanticBase
class ArticleRead(PydanticBase):
id: uuid.UUID
created_at: datetime.datetime
title: str
status: str
published: bool
category_id: uuid.UUID | None
```
### Routes
With the CRUD factory declared, routes become thin wrappers. Each route uses [ArticleCrud.filter_params()](https://fastapi-toolsets.d3vyce.fr/module/crud/#faceted-search) and [ArticleCrud.order_params()](https://fastapi-toolsets.d3vyce.fr/module/crud/#sorting) as FastAPI dependencies — these automatically generate the right query parameters based on the `facet_fields` and `order_fields` we declared.
#### Offset pagination
```python
@router.get("/offset")
async def list_articles_offset(
session: SessionDep,
filter_by: Annotated[dict[str, list[str]], Depends(ArticleCrud.filter_params())],
order_by: Annotated[
OrderByClause | None,
Depends(ArticleCrud.order_params(default_field=Article.created_at)),
],
page: int = Query(1, ge=1),
items_per_page: int = Query(20, ge=1, le=100),
search: str | None = None,
) -> OffsetPaginatedResponse[ArticleRead]:
return await ArticleCrud.offset_paginate(
session=session,
page=page,
items_per_page=items_per_page,
search=search,
filter_by=filter_by or None,
order_by=order_by,
schema=ArticleRead,
)
```
![Offset endpoint](img/image-1.webp)
**Example request:**
```
GET /articles/offset?page=2&items_per_page=2&search=fastapi&filter_by[status]=published&order_by=created_at&order_dir=desc
```
**Example response:**
```json
{
"items": [
{
"id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
"created_at": "2026-03-15T10:22:00Z",
"title": "Getting started with FastAPI",
"status": "published",
"published": true,
"category_id": "f1e2d3c4-b5a6-7890-abcd-ef0987654321"
},
{
"id": "b2c3d4e5-f6a7-8901-bcde-f01234567891",
"created_at": "2026-03-10T08:14:00Z",
"title": "FastAPI dependency injection explained",
"status": "published",
"published": true,
"category_id": "f1e2d3c4-b5a6-7890-abcd-ef0987654321"
}
],
"pagination": {
"total_count": 47,
"items_per_page": 2,
"page": 2,
"has_more": true
},
"pagination_type": "offset",
"filter_attributes": {
"status": ["draft", "published", "archived"],
"category__name": ["Python", "DevOps", "Architecture"]
}
}
```
#### Cursor pagination
```python
@router.get("/cursor")
async def list_articles_cursor(
session: SessionDep,
filter_by: Annotated[dict[str, list[str]], Depends(ArticleCrud.filter_params())],
order_by: Annotated[
OrderByClause | None,
Depends(ArticleCrud.order_params(default_field=Article.created_at)),
],
cursor: str | None = None,
items_per_page: int = Query(20, ge=1, le=100),
search: str | None = None,
) -> CursorPaginatedResponse[ArticleRead]:
return await ArticleCrud.cursor_paginate(
session=session,
cursor=cursor,
items_per_page=items_per_page,
search=search,
filter_by=filter_by or None,
order_by=order_by,
schema=ArticleRead,
)
```
![Cursor endpoint](img/image-2.webp)
**Example request (first page):**
```
GET /articles/cursor?items_per_page=2&search=fastapi&filter_by[status]=published
```
**Example response:**
```json
{
"items": [
{
"id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
"created_at": "2026-03-15T10:22:00Z",
"title": "Getting started with FastAPI",
"status": "published",
"published": true,
"category_id": "f1e2d3c4-b5a6-7890-abcd-ef0987654321"
},
{
"id": "b2c3d4e5-f6a7-8901-bcde-f01234567891",
"created_at": "2026-03-10T08:14:00Z",
"title": "FastAPI dependency injection explained",
"status": "published",
"published": true,
"category_id": "f1e2d3c4-b5a6-7890-abcd-ef0987654321"
}
],
"pagination": {
"next_cursor": "eyJjcmVhdGVkX2F0IjogIjIwMjYtMDMtMTBUMDg6MTQ6MDBaIn0=",
"prev_cursor": null,
"items_per_page": 2,
"has_more": true
},
"pagination_type": "cursor",
"filter_attributes": {
"status": ["draft", "published", "archived"],
"category__name": ["Python", "DevOps", "Architecture"]
}
}
```
Pass `next_cursor` as the `cursor` parameter to fetch the following page:
```
GET /articles/cursor?cursor=eyJjcmVhdGVkX2F0IjogIjIwMjYtMDMtMTBUMDg6MTQ6MDBaIn0=&items_per_page=2
```
#### Combined endpoint
You can also expose a single endpoint that supports both strategies via a `pagination_type` query parameter:
```python
@router.get("/")
async def list_articles(
session: SessionDep,
filter_by: Annotated[dict[str, list[str]], Depends(ArticleCrud.filter_params())],
order_by: Annotated[
OrderByClause | None,
Depends(ArticleCrud.order_params(default_field=Article.created_at)),
],
pagination_type: PaginationType = PaginationType.OFFSET,
page: int = Query(1, ge=1),
cursor: str | None = None,
items_per_page: int = Query(20, ge=1, le=100),
search: str | None = None,
) -> PaginatedResponse[ArticleRead]:
return await ArticleCrud.paginate(
session,
pagination_type=pagination_type,
page=page,
cursor=cursor,
items_per_page=items_per_page,
search=search,
filter_by=filter_by or None,
order_by=order_by,
schema=ArticleRead,
)
```
![Combined endpoint](img/image-3.webp)
The response shape adapts to the chosen strategy. With `pagination_type=offset` (default):
```
GET /articles/?pagination_type=offset&page=1&items_per_page=2&filter_by[status]=published
```
```json
{
"items": [...],
"pagination": {
"total_count": 47,
"items_per_page": 2,
"page": 1,
"has_more": true
},
"pagination_type": "offset",
"filter_attributes": {
"status": ["draft", "published", "archived"],
"category__name": ["Python", "DevOps", "Architecture"]
}
}
```
With `pagination_type=cursor`:
```
GET /articles/?pagination_type=cursor&items_per_page=2&filter_by[status]=published
```
```json
{
"items": [...],
"pagination": {
"next_cursor": "eyJjcmVhdGVkX2F0IjogIjIwMjYtMDMtMTBUMDg6MTQ6MDBaIn0=",
"prev_cursor": null,
"items_per_page": 2,
"has_more": true
},
"pagination_type": "cursor",
"filter_attributes": {
"status": ["draft", "published", "archived"],
"category__name": ["Python", "DevOps", "Architecture"]
}
}
```
## Conclusion
`fastapi-toolsets` removes the boilerplate of writing pagination, filtering and sorting from scratch every time. The [CrudFactory](https://fastapi-toolsets.d3vyce.fr/module/crud/#creating-a-crud-class) declaration is the single source of truth for what your API exposes — the routes just call it.
- Documentation: https://fastapi-toolsets.d3vyce.fr
- Source code: https://github.com/d3vyce/fastapi-toolsets
- Example code: https://github.com/d3vyce/fastapi-toolsets/tree/main/docs_src/examples/pagination_search