Зависимости в FastAPI: Depends на практике

Зачем вообще Depends, если можно импортировать функцию

Прямой импорт и вызов функции внутри обработчика работает — до первого теста. Чтобы подменить соединение с базой, подставить фейковый токен или отключить внешний сервис, придётся патчить модуль, менять глобальные переменные или городить фабрики вручную. Depends решает эту задачу на уровне фреймворка: зависимость объявляется в сигнатуре, FastAPI сам вызывает её перед обработчиком, прокидывает результат и позволяет подменить зависимость одной строкой в тестах через app.dependency_overrides.

Но у Depends есть неочевидные свойства: кеширование внутри запроса, поддержка генераторов для управления ресурсами, вложенность произвольной глубины. Именно здесь начинаются и рабочие приёмы, и типичные ошибки.

Depends — это не магия, а контейнер зависимостей, встроенный прямо в систему маршрутизации. Он делает ровно то, что делают DI-контейнеры в Spring или .NET, но без XML и без классов-конфигураторов: всё описано в сигнатуре функции.

Материал написан для FastAPI 0.110+ и Python 3.11+. В более ранних версиях отдельные аспекты (например, типизация Annotated) работают иначе — сверяйтесь с официальной документацией.

Базовый паттерн: функция-зависимость

Самый частый случай — получить текущего пользователя из токена. Вместо того чтобы дублировать парсинг заголовка в каждом эндпоинте, выносим логику в зависимость.

# auth.py — FastAPI 0.110+, Python 3.11+
from typing import Annotated
from fastapi import Depends, Header, HTTPException, status

async def get_current_user(
    authorization: Annotated[str, Header()],
) -> dict:
    """Извлекает пользователя из заголовка Authorization."""
    scheme, _, token = authorization.partition(" ")
    if scheme.lower() != "bearer" or not token:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Невалидный заголовок Authorization",
        )
    # Здесь вызов реальной проверки токена — JWT, сессия, внешний сервис
    user = await verify_token(token)  # ваша функция верификации
    if user is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Пользователь не найден",
        )
    return user

# routes.py
from typing import Annotated
from fastapi import APIRouter, Depends
from .auth import get_current_user

router = APIRouter()

@router.get("/me")
async def read_me(
    user: Annotated[dict, Depends(get_current_user)],
) -> dict:
    return user

Проверка: отправьте GET-запрос на /me без заголовка — должен вернуться 422 (FastAPI сам валидирует обязательный Header). С невалидным токеном — 401. С валидным — тело пользователя.

Вложенные зависимости и кеширование

Зависимость может зависеть от другой зависимости. FastAPI строит граф и разрешает его автоматически. При этом по умолчанию одна и та же зависимость в рамках одного запроса вызывается один раз, а результат переиспользуется — это кеширование.

# db.py
from typing import AsyncGenerator, Annotated
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

# engine и session_factory создаются при старте приложения

async def get_session() -> AsyncGenerator[AsyncSession, None]:
    """Отдаёт сессию БД и закрывает её после запроса."""
    async with session_factory() as session:
        yield session

SessionDep = Annotated[AsyncSession, Depends(get_session)]

# services.py
from .db import SessionDep

async def get_user_repo(session: SessionDep) -> UserRepo:
    return UserRepo(session)

async def get_order_repo(session: SessionDep) -> OrderRepo:
    return OrderRepo(session)

Обе репозитории получат одну и ту же сессию в рамках запроса. Это значит, что изменения, сделанные через UserRepo, видны в OrderRepo без дополнительных усилий — и commit вызывается один раз.

Когда кеширование мешает

Если вам нужна новая сессия при каждом вызове зависимости (редкий случай — параллельные транзакции), передайте use_cache=False:

fresh_session: Annotated[AsyncSession, Depends(get_session, use_cache=False)]

На практике use_cache=False нужен редко. Чаще всего, если вы к нему потянулись, архитектура зависимостей стоит пересмотра: две транзакции на один запрос — почти всегда признак проблемы, а не решение.

Генераторы: yield-зависимости для управления ресурсами

Зависимость с yield — это аналог контекстного менеджера. Код до yield выполняется перед обработчиком, код после — при завершении запроса (даже если обработчик упал с исключением).

Пример с блокировкой:

import asyncio
from typing import AsyncGenerator

_lock = asyncio.Lock()

async def with_lock() -> AsyncGenerator[None, None]:
    """Гарантирует, что в критическую секцию входит один запрос."""
    async with _lock:
        yield

Важно: если код после yield бросит исключение, FastAPI его залогирует, но клиент уже получит ответ. Не используйте cleanup-блок для бизнес-логики — только для освобождения ресурсов.

Подмена зависимостей в тестах

Это одна из главных причин существования Depends. Без него для тестирования эндпоинта пришлось бы поднимать реальную базу или заниматься monkey-patching. С Depends — одна строка:

# test_routes.py — pytest + httpx
import pytest
from httpx import ASGITransport, AsyncClient
from app.main import app
from app.auth import get_current_user

async def fake_user() -> dict:
    return {"id": 42, "username": "test_user"}

@pytest.fixture()
def client():
    app.dependency_overrides[get_current_user] = fake_user
    yield
    app.dependency_overrides.clear()  # обязательно чистим

@pytest.mark.anyio
async def test_read_me(client):
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test",
    ) as ac:
        resp = await ac.get("/me")
    assert resp.status_code == 200
    assert resp.json()["username"] == "test_user"

Ключевой момент: dependency_overrides — это обычный словарь на уровне экземпляра app. Если вы не очистите его после теста, следующий тест получит фейковую зависимость. Используйте фикстуру с yield или addfinalize.

Классы как зависимости

Depends принимает любой callable. Класс с __init__ — тоже callable. Это удобно, когда зависимость параметризуется:

from fastapi import Depends, Query
from typing import Annotated

class Pagination:
    def __init__(
        self,
        offset: int = Query(0, ge=0),
        limit: int = Query(20, ge=1, le=100),
    ):
        self.offset = offset
        self.limit = limit

PaginationDep = Annotated[Pagination, Depends()]

@router.get("/items")
async def list_items(pagination: PaginationDep):
    # pagination.offset, pagination.limit уже провалидированы
    ...

Обратите внимание на Depends() без аргумента: FastAPI сам возьмёт тип из аннотации. Это работает, потому что Pagination — callable, а параметры __init__ содержат аннотации Query, которые FastAPI умеет разрешать.

Типичные ошибки

1. Забытый Depends

# Неправильно: функция вызывается вручную, DI не работает
@router.get("/bad")
async def bad_route():
    user = await get_current_user(...)  # подмена в тестах не сработает

Если зависимость вызвана вручную, FastAPI о ней не знает: ни кеширования, ни cleanup, ни подмены через overrides.

2. Синхронная зависимость с I/O

Синхронная функция-зависимость выполняется в threadpool (через run_in_executor). Если она содержит блокирующий I/O, пул потоков быстро исчерпается. Решение: делайте зависимости с I/O асинхронными.

3. Тяжёлая логика в зависимости глобального уровня

Зависимость, подключённая через app = FastAPI(dependencies=[Depends(heavy)]), будет вызываться на каждом запросе, включая health-check. Подключайте такие зависимости к конкретному роутеру, а не ко всему приложению.

4. yield-зависимость без try/finally

# Опасно: если обработчик упадёт до yield, ресурс не освободится
async def bad_resource():
    conn = await acquire_connection()
    yield conn
    await conn.close()  # не выполнится при исключении до yield

# Правильно:
async def good_resource():
    conn = await acquire_connection()
    try:
        yield conn
    finally:
        await conn.close()

Сравнение подходов к организации зависимостей

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

• Измерьте время выполнения ваших зависимостей: добавьте middleware, который логирует длительность resolve-фазы. Если зависимости занимают больше времени, чем обработчик, — это сигнал.
• Проверьте, что dependency_overrides очищается в каждом тесте. Случайно оставленная подмена — источник flaky-тестов, которые проходят поодиночке и падают пакетом.
• Посмотрите на Annotated-паттерн (доступен начиная с FastAPI 0.95): он позволяет объявить типалиас с зависимостью один раз и использовать его без повторения Depends в каждом обработчике — как SessionDep в примере выше.
• Для сложных графов зависимостей изучите раздел Testing Dependencies в документации FastAPI — там описаны edge-кейсы с вложенными overrides.

Сноска. DI (Dependency Injection) — паттерн проектирования, при котором объект не создаёт свои зависимости сам, а получает их снаружи. В FastAPI роль DI-контейнера выполняет сам фреймворк: он анализирует сигнатуру функции, находит параметры с Depends, разрешает граф и передаёт готовые значения в обработчик.