Как работают dependencies с yield и как гарантировать cleanup ресурсов?

Как работают dependencies с yield

Dependency с yield — это генераторная функция, которую FastAPI использует как контекстный менеджер. Код до yield выполняется перед вызовом endpoint, код после — после отправки ответа. Это гарантирует cleanup ресурсов даже при исключениях.

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

engine = create_async_engine("postgresql+asyncpg://user:pass@db/mydb")
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        try:
            yield session          # <-- FastAPI передаёт сессию в endpoint
            await session.commit() # <-- выполняется после успешного ответа
        except Exception:
            await session.rollback()  # <-- откат при любом исключении
            raise

Внутренняя механика

FastAPI оборачивает генераторную dependency в contextlib.asynccontextmanager (для async) или contextlib.contextmanager (для sync). При исключении в endpoint FastAPI вызывает generator.athrow(exc), что позволяет блоку except в dependency получить это исключение.

Цепочки dependencies

async def get_redis():
    client = await aioredis.from_url("redis://redis:6379")
    try:
        yield client
    finally:
        await client.close()  # finally гарантирует закрытие даже при исключении

async def get_cache_service(
    redis = Depends(get_redis),
    db: AsyncSession = Depends(get_db),
):
    service = CacheService(redis=redis, db=db)
    yield service
    # cleanup redis и db выполняется в их собственных dependencies

Cleanup при background tasks

Важно: cleanup dependency выполняется до завершения BackgroundTask. Если фоновая задача использует ресурс из dependency (например, db-сессию), ресурс уже будет закрыт.

@app.post("/report")
async def generate_report(
    background_tasks: BackgroundTasks,
    db: AsyncSession = Depends(get_db),  # сессия закроется ДО выполнения задачи
):
    # Правильно: передавайте данные, а не саму сессию
    report_id = await db.execute(insert(Report).returning(Report.id))
    background_tasks.add_task(process_report, report_id.scalar())  # только ID
    return {"report_id": report_id.scalar()}

Тестирование dependency override

import pytest
from httpx import AsyncClient, ASGITransport
from main import app
from app.api.v1.deps import get_db

@pytest.fixture
async def test_db():
    # тестовая in-memory SQLite или отдельная тестовая БД
    async with test_session() as session:
        yield session

@pytest.mark.asyncio
async def test_create_user(test_db):
    app.dependency_overrides[get_db] = lambda: test_db
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        resp = await client.post("/users", json={"email": "a@b.com"})
    assert resp.status_code == 201
    app.dependency_overrides.clear()

Подводные камни

• Exception в коде до yield — если исключение произошло до yield, код после yield не выполнится. Ресурс не будет закрыт, если он был открыт до исключения. Используйте try/finally вокруг открытия ресурса.
• Исключение в блоке finally — если в finally бросается новое исключение, оригинальное теряется. Логируйте ошибки cleanup, не перебрасывайте их.
• Sync dependency с yield в async endpoint — FastAPI запускает sync generator в threadpool; если внутри используется async-код, будет ошибка «coroutine was never awaited». Используйте async def для async ресурсов.
• Кеширование dependencies — FastAPI кеширует dependency в рамках одного запроса; если один и тот же Depends используется в двух местах, ресурс создаётся один раз. Это экономит соединения, но может удивить при тестировании.
• BackgroundTasks + сессия БД — сессия из dependency закрывается до старта фоновой задачи; передавайте только примитивные данные (ID, строки), а не ORM-объекты.
• Утечка при unhandled exception в middleware — если кастомный middleware перехватывает исключение и не пробрасывает его дальше, FastAPI не вызовет cleanup dependency.

• Отвечать определением без production-сценария.
• Не называть runtime boundary, security boundary или failure mode.
• Игнорировать версию API, observability и тестовую проверку.

• Объясняет механизм своими словами и без выдуманных API.
• Называет реальные риски, диагностику и критерий корректности.
• Связывает ответ с текущей документацией и миграционными ограничениями.