FastAPI Pagination & Filtering

🎓 What You Will Learn

Limit-Offset Pagination: Simple pagination for most use cases
Cursor Pagination: Efficient pagination for large datasets
Filtering Query Parameters: Add filters to your API endpoints
Sorting Strategies: Multi-field sorting (ascending/descending)
Search Implementation: Full-text and fuzzy search patterns
Database Optimization: Indexing and query optimization strategies

FastAPISQLModelPostgreSQL

1Why Pagination Matters for Performance

Pagination is essential for building scalable APIs. Returning all records at once causes:

Memory issues: Large datasets consume excessive server memory
Slow response times: Clients wait for huge payloads to download
Poor user experience: Pages load slowly with unnecessary data
Database strain: Fetching all records stresses the database

Rule of Thumb: Always paginate results. Return 10-50 items per page for most use cases.

2Implementing Limit-Offset Pagination

Limit-offset is the most common pagination pattern. limit controls how many items to return. offset controls how many to skip.

app/routes/employees.py

from fastapi import APIRouter, Depends, Query
from sqlmodel import Session, select
from app.models import Employee
from app.database import get_session

router = APIRouter()

@router.get("/employees")
def list_employees(
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=100),
    session: Session = Depends(get_session)
):
    employees = session.exec(
        select(Employee).offset(skip).limit(limit)
    ).all()
    return {"data": employees, "skip": skip, "limit": limit}

📊 Interactive Pagination Demo

Showing 1-3 of 15 items

Employee 1Sales

Employee 2Engineering

Employee 3Marketing

Page 1 / 5

✅ Pagination keeps your API fast by returning only what's needed!

3Cursor-Based Pagination for Large Datasets

With offset, the database still scans all the skipped rows. That gets slow on big tables. Cursor pagination skips that work. It uses a bookmark (the cursor) from the last item and jumps straight to it.

app/routes/employees.py

@router.get("/employees/cursor")
def list_employees_cursor(
    cursor: int | None = Query(None),
    limit: int = Query(10, ge=1, le=100),
    session: Session = Depends(get_session)
):
    query = select(Employee).order_by(Employee.id)

    if cursor is not None:
        query = query.where(Employee.id > cursor)

    employees = session.exec(query.limit(limit + 1)).all()

    has_next = len(employees) > limit
    if has_next:
        employees = employees[:limit]

    next_cursor = employees[-1].id if employees and has_next else None

    return {
        "data": employees,
        "next_cursor": next_cursor,
        "has_next": has_next
    }

4Adding Filters to Query Parameters

Filters let clients request only the data they need. Common filters include exact match, range, and status. Update your /employees endpoint to accept filters.

app/routes/employees.py

@router.get("/employees")
def list_employees(
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=100),
    department: str | None = Query(None),
    salary_min: float | None = Query(None),
    salary_max: float | None = Query(None),
    session: Session = Depends(get_session)
):
    query = select(Employee)

    if department is not None:
        query = query.where(Employee.department == department)
    if salary_min is not None:
        query = query.where(Employee.salary >= salary_min)
    if salary_max is not None:
        query = query.where(Employee.salary <= salary_max)

    employees = session.exec(
        query.offset(skip).limit(limit)
    ).all()

    return {"data": employees}

Why check is not None? A value like 0 is valid but falsy in Python. if salary_min: would skip a filter of 0. is not None only skips missing values.

5Implementing Sorting

Allow clients to sort by different fields in ascending or descending order.

app/routes/employees.py

from enum import Enum
from sqlalchemy import asc, desc

class SortOrder(str, Enum):
    asc = "asc"
    desc = "desc"

# Only allow sorting by known columns
ALLOWED_SORT_FIELDS = {"id", "name", "department", "salary"}

@router.get("/employees")
def list_employees(
    skip: int = Query(0),
    limit: int = Query(10),
    sort_by: str = Query("id"),
    sort_order: SortOrder = Query(SortOrder.asc),
    session: Session = Depends(get_session)
):
    query = select(Employee)

    # Get the column to sort by (fall back to id)
    if sort_by not in ALLOWED_SORT_FIELDS:
        sort_by = "id"
    sort_column = getattr(Employee, sort_by)

    # Apply sort order
    if sort_order == SortOrder.desc:
        query = query.order_by(desc(sort_column))
    else:
        query = query.order_by(asc(sort_column))

    employees = session.exec(query.offset(skip).limit(limit)).all()
    return {"data": employees}

6Implementing Search Functionality

Search helps users find data by keyword. We use a simple case-insensitive ILIKE query here. For real full-text search on big data, use PostgreSQL tsvector or Elasticsearch.

app/routes/employees.py

@router.get("/employees/search")
def search_employees(
    q: str = Query(..., min_length=1),
    skip: int = Query(0),
    limit: int = Query(10),
    session: Session = Depends(get_session)
):
    # % matches any text before or after the keyword
    search_term = f"%{q}%"

    employees = session.exec(
        select(Employee)
        .where(
            Employee.name.ilike(search_term) |
            Employee.email.ilike(search_term)
        )
        .offset(skip)
        .limit(limit)
    ).all()

    return {"data": employees, "query": q}

7Database Optimization with Indexing

Indexes dramatically speed up filtering and searching. Create indexes on columns you frequently filter or sort by.

app/models.py

from sqlmodel import SQLModel, Field
from sqlalchemy import Index

class Employee(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)  # Index for searches
    email: str = Field(index=True, unique=True)
    department: str = Field(index=True)  # Index for filtering
    salary: float

    __table_args__ = (
        Index("idx_dept_salary", "department", "salary"),
    )

Index Tradeoff: Indexes speed up queries but slow down writes (INSERT/UPDATE). Use strategically on columns you query frequently.

8Creating Reusable Response Schemas

Define response schemas for pagination to keep your API consistent.

app/schemas/pagination.py

from typing import Generic, TypeVar, List
from pydantic import BaseModel, ConfigDict

T = TypeVar("T")

class PaginatedResponse(BaseModel, Generic[T]):
    data: List[T]
    skip: int
    limit: int
    total: int

class EmployeeResponse(BaseModel):
    # Lets Pydantic read data from database objects
    model_config = ConfigDict(from_attributes=True)

    id: int
    name: str
    email: str
    department: str

# Usage in route
@router.get("/employees", response_model=PaginatedResponse[EmployeeResponse])
def list_employees(
    skip: int = Query(0),
    limit: int = Query(10),
    session: Session = Depends(get_session)
):
    employees = session.exec(
        select(Employee).offset(skip).limit(limit)
    ).all()

    # Simple but slow: this loads every row just to count.
    # The next section shows a faster way.
    total = len(session.exec(select(Employee)).all())

    return PaginatedResponse(
        data=[EmployeeResponse.model_validate(e) for e in employees],
        skip=skip,
        limit=limit,
        total=total
    )

9Optimizing COUNT Queries

Counting rows in Python is slow. Let the database count for you with func.count.

app/repositories/employee_repository.py

from sqlalchemy import func
from sqlmodel import Session, select

from app.models import Employee

class EmployeeRepository:
    def __init__(self, session: Session):
        self.session = session

    def get_count(self) -> int:
        return self.session.exec(
            select(func.count(Employee.id))
        ).one()

    def get_paginated(self, skip: int, limit: int):
        return self.session.exec(
            select(Employee).offset(skip).limit(limit)
        ).all()

10Caching Paginated Results

Cache common pagination queries to reduce database load. Here is a simple time-based cache.

app/routes/employees.py

import time

# Simple in-memory cache: {key: (saved_at, data)}
_cache: dict = {}
CACHE_TTL = 60  # seconds

@router.get("/employees/cached")
def list_employees_cached(
    skip: int = Query(0),
    limit: int = Query(10),
    session: Session = Depends(get_session)
):
    key = f"employees:{skip}:{limit}"
    cached = _cache.get(key)

    if cached and time.time() - cached[0] < CACHE_TTL:
        return {"data": cached[1], "cached": True}

    employees = session.exec(
        select(Employee).offset(skip).limit(limit)
    ).all()
    _cache[key] = (time.time(), employees)

    return {"data": employees, "cached": False}

Careful with caching: Cached data can be stale. Do not use @lru_cache with database sessions — it never expires. For real apps, use Redis with a short TTL.

11API Documentation with OpenAPI

FastAPI automatically generates OpenAPI docs showing all your pagination and filtering parameters.

Automatic Documentation: Visit /docs in your browser to see interactive API documentation with pagination parameters.

12Testing Pagination and Filtering

Test all pagination scenarios including edge cases.

tests/test_employees.py

from fastapi.testclient import TestClient

from app.main import app

client = TestClient(app)

def test_pagination():
    response = client.get("/employees?skip=0&limit=10")
    assert response.status_code == 200
    assert len(response.json()["data"]) <= 10

def test_filter_department():
    response = client.get("/employees?department=Engineering")
    assert response.status_code == 200
    for employee in response.json()["data"]:
        assert employee["department"] == "Engineering"

def test_search():
    response = client.get("/employees/search?q=john")
    assert response.status_code == 200

13Common Pagination Pitfalls

Pitfall	Solution
Allowing unlimited limits	Set maximum limit (e.g., le=100 in Query)
No default pagination	Always paginate results
Inconsistent sorting	Use deterministic sorts (add ID as tiebreaker)
Missing indexes	Create indexes on filter/sort columns
No validation	Validate skip and limit parameters

14Advanced Topics

Once you master basic pagination:

Keyset Pagination: For APIs with constantly changing data
Elasticsearch: For advanced full-text search on large datasets
GraphQL: Alternative query language with built-in pagination
RabbitMQ/Kafka: For streaming results instead of pagination

15Resources & What's Next

You now understand pagination, filtering, and sorting—essential skills for building scalable APIs. Apply these patterns to every endpoint that returns lists.

Next Topics: Background tasks, long-running operations, caching strategies, and API rate limiting.

Congratulations! Your API is now optimized for performance and user experience. Keep pagination parameters consistent across all endpoints! 🚀

FastAPI Pagination & Filtering — Query Parameters & Database Optimization

🎓 What You Will Learn

2Implementing Limit-Offset Pagination

📊 Interactive Pagination Demo

4Adding Filters to Query Parameters

5Implementing Sorting

6Implementing Search Functionality

7Database Optimization with Indexing

8Creating Reusable Response Schemas

9Optimizing COUNT Queries

10Caching Paginated Results

11API Documentation with OpenAPI

12Testing Pagination and Filtering

13Common Pagination Pitfalls

14Advanced Topics

15Resources & What's Next

About the Author

Thirdy Gayares