Пагинация в FastAPI: готовое решение для любых эндпоинтов

Зачем нужна единая схема пагинации

FastAPI не навязывает способ разбиения ответа на страницы — в отличие от Django REST Framework, где PageNumberPagination работает из коробки. В итоге в одном проекте часть эндпоинтов возвращает {"items": [...], "total": 42}, часть — голый список, а третья — курсорную ссылку в заголовке. Клиент вынужден обрабатывать каждый формат отдельно, а Swagger-документация превращается в лоскутное одеяло.

Ниже — решение, которое закрывает 90% случаев offset-пагинации: единый формат ответа, автоматическая генерация OpenAPI-схемы и подключение через Depends. Всё написано для FastAPI 0.110+ и Pydantic v2. Если вам нужна курсорная пагинация (бесконечный скролл, real-time ленты) — это отдельная тема, здесь только offset/limit.

Что понадобится до начала

• Python 3.10+
• FastAPI 0.110+ с Pydantic v2 (pydantic ≥ 2.0)
• Любая async-совместимая ORM или голой asyncpg / databases — примеры даны для SQLAlchemy 2.x с async-сессией, но схема не зависит от ORM

Шаг 1. Параметры пагинации через Depends

Первое, что стоит вынести — параметры page и size. Объявляем их как зависимость, чтобы не дублировать в каждом роуте.

# app/pagination.py  (FastAPI 0.110+, Pydantic v2)
from fastapi import Query
from dataclasses import dataclass


@dataclass
class PaginationParams:
    """Параметры пагинации, внедряются через Depends."""
    page: int = Query(1, ge=1, description="Номер страницы, начиная с 1")
    size: int = Query(20, ge=1, le=100, description="Элементов на странице")

    @property
    def offset(self) -> int:
        return (self.page - 1) * self.size

    @property
    def limit(self) -> int:
        return self.size

@dataclass здесь — не случайность. FastAPI умеет резолвить dataclass-зависимости и подхватывает аннотации Query для OpenAPI. Если используете Pydantic-модель, понадобится кастомная __init__ или функция-фабрика — dataclass проще.

Проверка: после подключения в любой роут (pagination: PaginationParams = Depends()) в Swagger появятся два query-параметра — page и size — с валидацией и описанием.

Шаг 2. Обёртка ответа — дженерик Pydantic-модель

Клиенту нужен единый конверт: список элементов плюс метаданные для отрисовки пагинатора. Дженерик-модель Pydantic v2 решает задачу без дублирования.

# app/pagination.py  (продолжение)
from typing import Generic, TypeVar, Sequence
from pydantic import BaseModel
import math

T = TypeVar("T")


class PageResponse(BaseModel, Generic[T]):
    """Универсальный ответ с пагинацией."""
    items: list[T]
    total: int
    page: int
    size: int
    pages: int  # общее число страниц

    @classmethod
    def create(
        cls,
        items: Sequence[T],
        total: int,
        params: PaginationParams,
    ) -> "PageResponse[T]":
        return cls(
            items=list(items),
            total=total,
            page=params.page,
            size=params.size,
            pages=math.ceil(total / params.size) if params.size else 0,
        )

Теперь PageResponse[UserOut] в аннотации роута генерирует в OpenAPI полную JSON-схему с конкретными полями UserOut внутри items. Фронтенд-генераторы типов (openapi-typescript, orval) подхватят её автоматически.

Почему classmethod, а не __init__

Pydantic v2 валидирует данные в __init__. Перегрузка конструктора ради вычисления pages ломает валидацию или требует model_validator. Фабричный метод create проще, очевиднее и не конфликтует с сериализацией.

Шаг 3. Подключение к роуту

# app/routes/users.py
from fastapi import APIRouter, Depends
from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession

from app.database import get_session
from app.models import User
from app.schemas import UserOut
from app.pagination import PaginationParams, PageResponse

router = APIRouter()


@router.get("/users", response_model=PageResponse[UserOut])
async def list_users(
    pagination: PaginationParams = Depends(),
    session: AsyncSession = Depends(get_session),
):
    # Общее число записей
    total = await session.scalar(
        select(func.count()).select_from(User)
    )

    # Страница данных
    rows = await session.scalars(
        select(User)
        .offset(pagination.offset)
        .limit(pagination.limit)
        .order_by(User.id)
    )

    return PageResponse.create(
        items=rows.all(),
        total=total,
        params=pagination,
    )

Проверка: запрос GET /users?page=2&size=10 возвращает JSON с полями items, total, page, size, pages. Swagger показывает схему с вложенным UserOut.

Что получилось: формат ответа

Любая пагинация — компромисс. Offset дешёв в реализации, но при большом offset база всё равно читает и отбрасывает предыдущие строки. Если у вас таблица на миллионы записей и пользователи реально листают дальше сотой страницы — переходите на keyset-пагинацию. Для большинства CRUD-приложений offset работает без проблем.

Фильтрация и сортировка рядом с пагинацией

Типичная ошибка — засунуть фильтры внутрь PaginationParams. Фильтры специфичны для сущности, пагинация — нет. Держите их отдельными зависимостями и комбинируйте в роуте:

@router.get("/orders", response_model=PageResponse[OrderOut])
async def list_orders(
    status: str | None = Query(None),
    pagination: PaginationParams = Depends(),
    session: AsyncSession = Depends(get_session),
):
    query = select(Order)
    count_query = select(func.count()).select_from(Order)

    if status:
        query = query.where(Order.status == status)
        count_query = count_query.where(Order.status == status)

    total = await session.scalar(count_query)
    rows = await session.scalars(
        query.offset(pagination.offset)
        .limit(pagination.limit)
        .order_by(Order.created_at.desc())
    )
    return PageResponse.create(rows.all(), total, pagination)

Два запроса к базе — count и select — неизбежны при offset-пагинации. Если count тормозит, кешируйте его отдельно или отдавайте total: null при запросах с фильтрами, а на фронте скрывайте номер последней страницы.

Если не получилось

Swagger показывает items как list[object] без полей

Убедитесь, что response_model=PageResponse[UserOut] указан явно. Без аннотации response_model FastAPI берёт тип из return-аннотации, а если её нет — не может вывести дженерик.

ValidationError: Input should be a valid dictionary

Вы передали в items ORM-объекты, а Pydantic v2 не умеет читать их атрибуты без model_config = ConfigDict(from_attributes=True) в схеме UserOut. Добавьте эту строку в вашу Pydantic-модель.

page=0 или size=0 проходят валидацию

Проверьте, что используете Query(1, ge=1), а не просто значение по умолчанию. Без ge FastAPI не генерирует ограничение.

COUNT(*) на большой таблице работает секунды

Это не баг пагинации, а особенность PostgreSQL: COUNT(*) по таблице без WHERE требует seq scan. Варианты: кешировать счётчик в Redis с TTL, использовать приблизительный reltuples из pg_class, или отказаться от показа общего числа страниц.

Сноска. Offset-пагинация — способ разбиения результатов, при котором клиент передаёт номер страницы (или смещение), а база пропускает первые N строк. Альтернатива — keyset-пагинация (она же cursor-based), где вместо смещения передаётся значение последнего поля сортировки; она эффективнее на больших таблицах, но сложнее в реализации и не позволяет перейти на произвольную страницу.

Что делать дальше

• Добавьте тесты: проверяйте граничные случаи — запрос страницы за пределами pages, size=1, пустая таблица. PageResponse.create с пустым списком и total=0 должен возвращать pages=0.
• Если эндпоинтов больше десяти — вынесите формирование count + select в отдельную функцию-хелпер репозитория, чтобы не дублировать два запроса в каждом роуте.
• Замерьте время COUNT на продовых объёмах. Если оно выходит за SLA — это сигнал перейти на keyset или кешированный счётчик, а не повод добавлять OFFSET с индексом.
• Ознакомьтесь с библиотекой fastapi-pagination — она реализует похожую идею с большим набором стратегий (cursor, limit-offset, page-number). Если ваш проект небольшой, собственное решение из статьи проще и прозрачнее; если пагинация нужна в десятках вариантов — библиотека сэкономит время.