FastAPI dependency injection: рабочие паттерны и антипаттерны
FastAPI часто хвалят за систему зависимостей. И за дело: Depends — это, наверное, главная фича, которая делает фреймворк не просто «асинхронным Flask», а полноценным инструментом. Но чем больше команда, тем чаще видишь одни и те же ошибки в работе с DI, которые потом выливаются в утечки соединений, дублирование кода и невозможность нормально протестировать ручки.
Я расскажу про паттерны, которые у нас прижились на нескольких сервисах, и про антипаттерны, которые мы вычистили после боли на проде.
Чем DI в FastAPI отличается от классического
В Spring или .NET DI-контейнер сидит сбоку и сам решает, кому что подсунуть. В FastAPI всё иначе: зависимости — это просто функции, которые ты явно объявляешь параметрами эндпоинта. Никакого скрытого реестра, никакой магии. Это ближе к функциональному подходу, и в этом сила: ты всегда видишь, что именно получает функция.
База выглядит так:
from fastapi import Depends, FastAPI
from sqlalchemy.ext.asyncio import AsyncSession
app = FastAPI()
async def get_session() -> AsyncSession:
async with async_session_factory() as session:
yield session
@app.get("/users/{user_id}")
async def read_user(user_id: int, session: AsyncSession = Depends(get_session)):
return await session.get(User, user_id)Дальше всё крутится вокруг этой механики: функция get_session — это зависимость, и FastAPI вызывает её перед каждым обработчиком, а после ответа корректно закрывает через yield.
Паттерн: фабрики зависимостей
Часто нужна не просто зависимость, а зависимость с параметрами. Например, проверка прав: один эндпоинт требует роль admin, другой — editor. Решение — функция, возвращающая зависимость.
from fastapi import HTTPException, Depends
def require_role(role: str):
async def checker(user: User = Depends(get_current_user)) -> User:
if role not in user.roles:
raise HTTPException(403, "forbidden")
return user
return checker
@app.delete("/posts/{post_id}")
async def delete_post(post_id: int, _: User = Depends(require_role("admin"))):
...Замыкание над параметром role позволяет переиспользовать общую логику. Это работает чисто, без классов и без магических декораторов. Тесты на такие эндпоинты пишутся легко — ты просто переопределяешь get_current_user.
Паттерн: сессия БД в зависимостях, не в репозитории
Я регулярно вижу такую конструкцию:
class UserRepository:
def __init__(self):
self.session = async_session_factory()
async def get(self, user_id: int):
return await self.session.get(User, user_id)Сессия живёт в репозитории и закрывается непонятно когда. Потом начинаются вопросы: почему изменения не видны, почему пул соединений утекает, откуда лишние транзакции. Правильный путь — отдавать сессию репозиторию извне.
class UserRepository:
def __init__(self, session: AsyncSession):
self.session = session
async def get(self, user_id: int) -> User | None:
return await self.session.get(User, user_id)
async def get_user_repo(session: AsyncSession = Depends(get_session)) -> UserRepository:
return UserRepository(session)
@app.get("/users/{user_id}")
async def read_user(user_id: int, repo: UserRepository = Depends(get_user_repo)):
return await repo.get(user_id)Жизнь сессии теперь привязана к запросу. Когда обработчик завершается, сессия закрывается, соединение возвращается в пул. Никаких утечек.
Паттерн: сервисный слой через DI
Если ручка делает что-то сложнее «достать запись из БД», логику стоит вынести в сервис. Сервис тоже даём через зависимости.
class OrderService:
def __init__(self, orders: OrderRepository, payments: PaymentClient):
self.orders = orders
self.payments = payments
async def create(self, data: OrderCreate) -> Order:
order = await self.orders.create(data)
await self.payments.charge(order)
return order
async def get_order_service(
orders: OrderRepository = Depends(get_order_repo),
payments: PaymentClient = Depends(get_payment_client),
) -> OrderService:
return OrderService(orders, payments)
@app.post("/orders")
async def create_order(
payload: OrderCreate,
service: OrderService = Depends(get_order_service),
):
return await service.create(payload)Эндпоинт стал тонким, бизнес-логика — в сервисе, репозитории и внешние клиенты — в их зависимостях. Всё это легко тестировать: в тестах подменяешь любую зависимость через app.dependency_overrides.
Паттерн: dependency_overrides для тестов
Это, пожалуй, главная киллер-фича DI в FastAPI. Тебе не нужны mock-патчи на уровне импортов — достаточно подменить функцию-зависимость.
def fake_payment_client():
class Fake:
async def charge(self, order):
return None
return Fake()
def test_create_order(client):
app.dependency_overrides[get_payment_client] = fake_payment_client
resp = client.post("/orders", json={"items": [...]})
assert resp.status_code == 201
app.dependency_overrides.clear()Никаких monkeypatch на третьесторонние клиенты. Никаких глобальных моков. Чистый и предсказуемый способ изолировать тест.
Антипаттерн: глобальные синглтоны
Иногда вижу такое:
redis_client = Redis(...)
@app.get("/cache/{key}")
async def get_cached(key: str):
return await redis_client.get(key)Технически работает. Но redis_client создан на импорте модуля — а значит, на момент импорта event loop ещё нет, и любой клиент, который заводит соединение в __init__, упадёт. Плюс ты не можешь подменить его в тестах. Делай через зависимость с yield или через lifespan-инициализацию.
from contextlib import asynccontextmanager
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.redis = Redis.from_url(REDIS_URL)
yield
await app.state.redis.close()
async def get_redis(request: Request) -> Redis:
return request.app.state.redisАнтипаттерн: тяжёлые операции в зависимостях
Зависимость вызывается на каждый запрос. Если ты в зависимости делаешь поход в БД, чтобы достать пользователя, потом ещё раз ходишь в БД за его правами, потом за фичфлагами — это всё на каждый запрос. Один эндпоинт может потянуть пять подзапросов до того, как ты вообще написал хоть одну строку основной логики.
Решение — кэшировать результат внутри запроса через Depends(..., use_cache=True). По умолчанию FastAPI и так кэширует одинаковые зависимости в рамках одного запроса, но если ты случайно делаешь несколько одинаковых функций — кэш не сработает. Делай одну общую функцию.
Антипаттерн: сложные классы вместо функций
Видел реализации типа class GetCurrentUser: def __call__(self, ...): ... с десятком методов. Класс как зависимость работает (FastAPI принимает любой callable), но почти всегда это переусложнение. Если у тебя в DI логика на 200 строк — это уже сервис, и он должен лежать отдельно, а зависимость должна просто его собрать.
Антипаттерн: Depends внутри функции
Иногда люди пытаются вызвать зависимость вручную:
async def get_data():
session = Depends(get_session) # так не работает
...FastAPI разрешает зависимости только в параметрах эндпоинтов и других зависимостей. Если тебе нужна зависимость глубоко в коде — это сигнал, что ты пытаешься обмануть систему. Передай зависимость как аргумент явно или подумай, нужен ли тебе вообще DI в этом месте.
Sub-dependencies и иерархия
Зависимости можно вкладывать сколько угодно. FastAPI разрешает граф автоматически и кеширует промежуточные значения в рамках одного запроса. Это позволяет строить чёткую иерархию: get_session → get_user_repo → get_user_service → эндпоинт. Каждый уровень знает только своих прямых соседей.
Главное — не делать циклов. Если две зависимости тянут друг друга — это сигнал переосмыслить, что у тебя сервис, а что репозиторий, и не залез ли один слой в другой.
Подытожим
Хорошая работа с DI в FastAPI выглядит примерно так: тонкие эндпоинты, чёткая лестница зависимостей от инфраструктуры (сессия, redis, http-клиент) к репозиториям и сервисам, явная подмена в тестах через dependency_overrides. Никаких глобальных синглтонов, никакой магии.
DI в FastAPI хорош ровно настолько, насколько хороши твои pydantic-модели. Если объявления зависимостей читаются как описание, что нужно эндпоинту — ты на правильном пути. Если зависимость превращается в кашу из проверок и побочных эффектов — выноси в сервис, а DI оставляй для сборки.