Зависимости вглубь: классы, yield, кэширование

Базовый Depends вы уже знаете — теперь разберём, как зависимости держат соединения с БД, переиспользуются в рамках запроса и подменяются в тестах.

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

Зачем это нужно: соединение с базой нельзя просто «получить» — его обязательно надо закрыть, иначе пул соединений исчерпается и приложение встанет. Зависимости с yield дают надёжное «открыл — поработал — закрыл» без дублирования кода в каждом эндпоинте. А классы-зависимости и кэширование убирают повторную работу и делают код тестируемым.

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

В Depends можно передать не только функцию, но и класс. Тогда FastAPI разберёт параметры __init__ как параметры запроса, создаст экземпляр и отдаст его в обработчик. Это удобно для группировки повторяющихся query-параметров:

from fastapi import Depends, FastAPI

app = FastAPI()

class Pagination:
    def __init__(self, page: int = 1, size: int = 20):
        self.page = page
        self.size = size
        self.offset = (page - 1) * size

@app.get("/users")
def list_users(pg: Pagination = Depends(Pagination)):
    return {"offset": pg.offset, "limit": pg.size}

Теперь ?page=3&size=10 автоматически превратится в объект Pagination с готовым offset. Поскольку тип параметра и сам класс совпадают, FastAPI разрешает короткую запись Depends() без аргумента — он возьмёт класс из аннотации.

Зависимости с yield для ресурсов БД

Классический паттерн — сессия БД, которую надо закрыть в любом случае:

from app.db import SessionLocal

def get_db():
    db = SessionLocal()      # код ДО yield: открыли соединение
    try:
        yield db             # отдали сессию в обработчик
    finally:
        db.close()           # код ПОСЛЕ yield: закрыли — всегда

@app.get("/users")
def list_users(db = Depends(get_db)):
    return db.query(User).all()

FastAPI выполнит код до yield, передаст db в обработчик, дождётся завершения запроса и затем выполнит часть после yield. Блок finally здесь обязателен: он гарантирует закрытие сессии даже если обработчик выбросил исключение. Это и есть промышленный способ управления соединениями — никаких утечек.

Под-зависимости

Зависимости выстраиваются в цепочки: одна зависимость может зависеть от другой. FastAPI разрешит весь граф автоматически:

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

def get_current_user(db = Depends(get_db), token: str = Header(...)):
    return db.query(User).filter_by(token=token).first()

def get_admin(user = Depends(get_current_user)):
    if not user.is_admin:
        raise HTTPException(403, "Только для админов")
    return user

@app.get("/admin/stats")
def stats(admin = Depends(get_admin)):
    return {"ok": True}

Чтобы выдать admin, FastAPI сначала получит db, затем current_user, затем проверит права — и всё это в правильном порядке, без ручной передачи аргументов.

Кэширование в рамках запроса

Заметьте: get_db в примере выше используется и напрямую, и внутри get_current_user. Будет ли создано два соединения? Нет. По умолчанию FastAPI кэширует результат зависимости в пределах одного запроса: если одна и та же зависимость встречается в графе несколько раз, она вызывается лишь однажды, а результат переиспользуется. Это критично — иначе тяжёлая зависимость (запрос к БД, чтение пользователя) выполнялась бы по нескольку раз за запрос.

Когда кэш мешает (например, нужно гарантированно получать свежий результат при каждом вызове), его отключают флагом use_cache=False:

def fresh_token():
    return secrets.token_hex(8)

@app.get("/x")
def handler(
    a: str = Depends(fresh_token),                       # из кэша
    b: str = Depends(fresh_token, use_cache=False),       # вызовется заново
):
    return {"a": a, "b": b}

Идею кэша «один вызов на запрос» легко показать на чистом Python:

cache = {}
calls = {"count": 0}

def resolve(dep, use_cache=True):
    if use_cache and dep in cache:
        return cache[dep]
    calls["count"] += 1          # реально выполняем зависимость
    value = f"{dep}-result"
    if use_cache:
        cache[dep] = value
    return value

# одна зависимость встречается в графе дважды
resolve("get_db")
resolve("get_db")                # взято из кэша, повторно не выполнялась
resolve("get_db", use_cache=False)  # принудительно заново

print("реальных вызовов get_db:", calls["count"])

Вывод:

реальных вызовов get_db: 2

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

Главная практическая выгода Depends — тестируемость. В тестах не нужна настоящая БД: зависимость подменяется на фейковую через словарь app.dependency_overrides, где ключ — оригинальная функция, значение — замена:

from fastapi.testclient import TestClient
from app.main import app
from app.dependencies import get_db

def fake_db():
    return FakeSession([{"id": 1, "name": "test"}])

app.dependency_overrides[get_db] = fake_db   # подменили БД на фейк

client = TestClient(app)

def test_list_users():
    resp = client.get("/users")
    assert resp.status_code == 200

# после теста вернуть как было:
app.dependency_overrides.clear()

Никакой магии: FastAPI перед вызовом каждой зависимости заглядывает в этот словарь и, найдя ключ, использует замену. Код эндпоинтов при этом не меняется ни на строку — в этом вся сила инъекции зависимостей.

Как это работает под капотом

На старте FastAPI строит для каждого эндпоинта граф зависимостей — дерево вызовов с разрешённым порядком. На каждый запрос создаётся отдельный кэш (обычный словарь), ключом в котором служит сама функция-зависимость (плюс её под-параметры). Перед вызовом зависимости FastAPI проверяет: есть ли уже значение в кэше этого запроса и не стоит ли use_cache=False — если значение есть, функция не вызывается повторно. Зависимости с yield кладутся в стек завершения (по сути, как контекстные менеджеры): после отправки ответа FastAPI «доматывает» каждый генератор за yield в порядке, обратном открытию, — поэтому сессия, открытая первой, закроется последней. Этот же стек ловит исключения, что и позволяет finally отрабатывать при ошибке в обработчике.

Частые ошибки

• Открыли ресурс в yield-зависимости, но не обернули в try/finally — при исключении в обработчике соединение не закроется, пул утечёт.
• Ставят настоящую работу с БД и ждут, что зависимость выполнится заново на каждое использование, забывая про кэш на уровне запроса (нужен use_cache=False).
• В dependency_overrides в качестве ключа кладут не ту функцию (например, обёртку вместо оригинала) — подмена молча не срабатывает, тест бьётся о реальную зависимость.
• Забыли app.dependency_overrides.clear() после теста — подмена «протекает» в соседние тесты и ломает их.
• Думают, что кэш зависимостей живёт между запросами, — нет, он создаётся заново на каждый запрос и не разделяется между ними.

Итоги

• Класс в Depends превращает свои параметры __init__ в параметры запроса и отдаёт готовый объект.
• Зависимость с yield + try/finally — промышленный способ открыть и гарантированно закрыть ресурс (сессию БД).
• Под-зависимости образуют граф, который FastAPI разрешает автоматически в правильном порядке.
• По умолчанию зависимость кэшируется в пределах запроса и вызывается один раз; отключается через use_cache=False.
• app.dependency_overrides[оригинал] = замена подменяет зависимости в тестах без правки кода эндпоинтов; после теста — clear().