FastAPI без магии: зависимости, жизненный цикл приложения и тонкая настройка валидации

FastAPI часто воспринимают как “быстрое магическое API”. На практике же он ценится не за магию, а за инженерную предсказуемость: декларативная модель зависимостей, формализованный жизненный цикл и сильная валидация через Pydantic. Если разобраться в механике, можно добиться двух вещей, которые в проде важнее любых “фичей”:

1. Корректное и безопасное управление ресурсами (соединения с БД, очереди, файловые дескрипторы, фоновые задачи).
2. Предсказуемая валидация входных данных (где и как именно валидировать, как формировать ошибки, как контролировать “молчаливые” изменения типов).

Ниже разберём, как устроены dependencies и lifespan, какие есть подводные камни и как сделать валидацию тонкой и стабильной. Примеры — рабочие и их можно адаптировать под реальный проект.

Dependencies: DI без “магии” и почему важно понимать, где создаются объекты

Что такое зависимость в FastAPI по сути

В FastAPI dependency — это функция (или callable), которую фреймворк вызывает, чтобы получить значение, нужное эндпоинту. В отличие от “классического” DI-контейнера, здесь всё завязано на сигнатуры Python-функций и анализ параметров:

• Если параметр объявлен через Depends(...), FastAPI понимает: нужно вызвать указанную зависимость.
• Типы и значения параметров зависимости — это ещё и путь к валидации.
• Сама зависимость может быть “попутчиком” (создаётся на каждый запрос) или “одиночкой” в рамках приложения — если вы аккуратно используете контекст жизненного цикла.

Зависимости на каждый запрос: удобно, но потенциально дорого

Типичный пример: создать сессию БД на запрос.

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

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/db"

engine = create_async_engine(DATABASE_URL, pool_pre_ping=True)
SessionLocal = async_sessionmaker(bind=engine, expire_on_commit=False)

async def get_db_session() -> AsyncSession:
    async with SessionLocal() as session:
        yield session

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: int, db: AsyncSession = Depends(get_db_session)):
    # db гарантированно живёт на протяжении обработки запроса
    # и закрывается после выхода из yield-контекста
    ...

Почему это безопасно: yield позволяет “обернуть” ресурс в контекст жизненного цикла конкретного запроса.

Почему это может быть дорого: если у вас тяжёлое создание объекта (например, клиент внешнего API с TLS-настройками, большой warm-up), то создание на каждый запрос может ухудшить latency и нагрузить систему.

“Мелкая ошибка”: зависимости как побочные эффекты

Частая проблема — когда зависимость выполняет побочную работу неявно. Например, логирует, пишет в очередь, открывает соединение и т. п., но забывают корректно завершить ресурс. Если вы видите зависимость, которая:

• не использует yield;
• не закрывает соединение в finally;
• не привязана к lifespan приложения,

то высок риск утечек.

Моделирование “синглтона” через зависимость, но с контролем

Например, у вас есть клиент векторного хранилища или external service, который лучше создать один раз при старте приложения. Здесь важно: создать в lifespan и передать через dependency.

FastAPI позволяет держать состояние приложения в app.state, а затем зависимость просто читает это состояние.

from fastapi import FastAPI, Depends
from typing import AsyncGenerator

class ExternalClient:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self._connected = False

    async def connect(self) -> None:
        # условно: инициализация сессии, handshake, загрузка метаданных
        self._connected = True

    async def close(self) -> None:
        self._connected = False

    async def ping(self) -> bool:
        return self._connected

async def get_client(app: FastAPI) -> ExternalClient:
    # Это пример: реальный вариант — зависимость через параметр request,
    # но для простоты покажем идею.
    return app.state.external_client

app = FastAPI()

@app.get("/health-ext")
async def health_ext(client: ExternalClient = Depends(lambda: None)):
    ...

Этот пример намеренно неполный, потому что в реальном коде зависимость должна получать доступ к request (а значит и request.app.state), либо получать app иначе. Практически используемый паттерн ниже.

Lifespan: безопасный жизненный цикл приложения и корректное освобождение ресурсов

Почему lifespan важнее, чем “побочные события” в обработчиках

lifespan — это механизм, который запускается при старте приложения и корректно завершает ресурсы при остановке. В проде это критично:

• чтобы не плодить соединения;
• чтобы не держать пул в “полузакрытом” состоянии;
• чтобы корректно завершить фоновые задачи;
• чтобы дождаться завершения flush/commit (если применимо).

Базовый шаблон lifespan для асинхронных ресурсов

В FastAPI lifespan реализуется как асинхронный контекст-менеджер.

from contextlib import asynccontextmanager
from fastapi import FastAPI

class ResourcePool:
    async def start(self) -> None:
        # прогрев, подключение, подготовка
        ...

    async def stop(self) -> None:
        # закрыть соединения/пул, дождаться завершений
        ...

@asynccontextmanager
async def lifespan(app: FastAPI):
    pool = ResourcePool()
    await pool.start()
    app.state.pool = pool

    try:
        yield
    finally:
        await pool.stop()

app = FastAPI(lifespan=lifespan)

Что здесь важно:

• В try/finally вы гарантируете освобождение ресурса даже при ошибке во время обслуживания.
• yield — точка, в которой приложение работает.
• Состояние хранится в app.state, и дальше вы можете аккуратно читать его из зависимостей.

Доступ к app.state из dependency

Паттерн: dependency принимает Request, а оттуда берёт request.app.state.

from fastapi import Request, Depends

async def get_pool(request: Request):
    return request.app.state.pool

@app.get("/do-something")
async def do_something(pool=Depends(get_pool)):
    ...

Тут нет “магии”: FastAPI передаёт нужный объект Request в dependency, а вы читаете поле. Такой подход хорошо масштабируется: можно держать несколько ресурсов (db, redis, client, feature flags) и отдавать их через зависимости.

Подводный камень: не смешивайте lifespan и request-scoped в одном месте

Если объект создан в lifespan и закрывается при остановке приложения, не закрывайте его в dependency на каждый запрос. И наоборот: если объект живёт только в рамках запроса (created per-request), не храните его в app.state.

Типичный анти-паттерн: создать клиент в request-scoped dependency, сохранить его в app.state “на всякий случай”. В итоге вы получите состояние гонки и закрытие объекта не в том месте.

Тонкая настройка валидации: где именно контролировать входные данные

Уровни валидации в FastAPI

Входные данные проходят несколько слоёв проверки:

1. Сигнатура эндпоинта — query params, path params, header params.
2. Модель тела запроса (Pydantic) — при парсинге JSON.
3. Дополнительная логика — валидаторы (@field_validator, @model_validator), кастомные проверки, бизнес-правила.
4. Обработка ошибок — как FastAPI преобразует ValidationError в HTTP-ответ.

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

Строгие типы и “скрытые” преобразования

Pydantic исторически допускает преобразование типов (например, строка "1" → int 1). В большинстве случаев это удобно, но иногда валидация становится “мягкой” и вы начинаете принимать данные, которые по контракту не должны проходить.

Если контракт API строгий, лучше:

• жёстко ограничивать поля;
• использовать StrictInt, StrictStr, StrictBool;
• добавлять constr, conint, condecimal и разумные ограничения.

Пример строгих типов:

from pydantic import BaseModel, Field, StrictInt, StrictStr

class CreateUserIn(BaseModel):
    user_id: StrictInt = Field(gt=0, description="Положительный ID пользователя")
    email: StrictStr
    display_name: StrictStr = Field(min_length=1, max_length=100)

Если клиент пришлёт "user_id": "123", это будет ошибка валидации (а не тихое преобразование).

Нормализация данных: делайте это осознанно

Иногда “строгость” не нужна, но нужна предсказуемость. Например, нормализовать email (lowercase) — логично. Делайте это в валидаторе, а не “в середине обработки” эндпоинта, чтобы ошибка (или нормализация) была воспроизводимой.

from pydantic import BaseModel, Field, EmailStr, field_validator

class CreateUserIn(BaseModel):
    email: EmailStr
    display_name: str = Field(min_length=1, max_length=100)

    @field_validator("email")
    @classmethod
    def normalize_email(cls, v: EmailStr) -> str:
        return v.lower()

Важно: если вы нормализуете поле, делайте это в модели, чтобы downstream-логика (слой сервиса) работала с уже приведёнными данными.

Сложная логика валидации: model_validator вместо “размазанных if”

Если валидация зависит от нескольких полей, используйте @model_validator(mode="after") (в Pydantic v2) или аналогичный подход.

from pydantic import BaseModel, Field, model_validator
from typing import Optional

class TransferIn(BaseModel):
    from_account: str
    to_account: str
    amount: float = Field(gt=0)
    memo: Optional[str] = Field(default=None, max_length=200)

    @model_validator(mode="after")
    def check_accounts(self):
        if self.from_account == self.to_account:
            raise ValueError("from_account и to_account не должны совпадать")
        if self.memo is None and self.amount > 10_000:
            # пример business-rule
            raise ValueError("Для крупных переводов требуется memo")
        return self

Здесь ошибки будут формироваться единообразно, а модель гарантирует целостность данных до попадания в endpoint.

Управление деталями ошибок: что видит клиент

По умолчанию FastAPI отдаёт структуру ошибок вида:

• loc: где ошибка (path/query/body/поле),
• msg: человекочитаемое сообщение,
• type: тип ошибки.

Чтобы сделать поведение предсказуемым для интеграторов, стоит:

1. Давать понятные сообщения в ValueError;
2. Не полагаться на “стандартные” тексты библиотек для бизнес-правил;
3. Следить за loc: если вы делаете собственные исключения, не ломайте формат.

Практический совет: для бизнес-валидации пишите свои сообщения, а для технических — используйте стандартные типы ошибок.

Режимы Pydantic: “как часто ломают контракт” через конфиги

Конфигурация Pydantic определяет, какие входные форматы допускаются. Например, что делать с лишними полями (extra), как обрабатывать преобразования и т. п.

Для стабильного контракта обычно выбирают:

• запрет лишних полей (чтобы клиент не присылал “мусор” и вы не игнорировали его молча);
• строгие типы там, где это важно.

Пример:

from pydantic import BaseModel, ConfigDict, StrictInt

class StrictBody(BaseModel):
    model_config = ConfigDict(extra="forbid")

    user_id: StrictInt

Если клиент пришлёт { "user_id": 1, "hack": true }, то будет ошибка. Это часто полезно в корпоративной среде, где требования к контракту жёсткие.

Практический сценарий: безопасные ресурсы + предсказуемая валидация

Соберём всё вместе: lifespan создаёт “пул” или клиент, зависимость отдаёт его эндпоинтам, а модель обеспечивает контракт.

Код примера

from contextlib import asynccontextmanager
from fastapi import FastAPI, Request, Depends, HTTPException
from pydantic import BaseModel, ConfigDict, StrictStr, StrictInt, Field, model_validator
from typing import Optional
import asyncio

class MailerClient:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self._ready = False

    async def start(self):
        # имитация подключения
        await asyncio.sleep(0.1)
        self._ready = True

    async def close(self):
        # имитация освобождения ресурсов
        await asyncio.sleep(0.05)
        self._ready = False

    async def send(self, to: str, subject: str, body: str):
        if not self._ready:
            raise RuntimeError("Mailer client is not ready")
        # имитация отправки
        await asyncio.sleep(0.05)
        return {"message_id": "abc123"}

@asynccontextmanager
async def lifespan(app: FastAPI):
    client = MailerClient(api_key="secret")
    await client.start()
    app.state.mailer = client
    try:
        yield
    finally:
        await client.close()

app = FastAPI(lifespan=lifespan)

async def get_mailer(request: Request) -> MailerClient:
    return request.app.state.mailer

class SendEmailIn(BaseModel):
    model_config = ConfigDict(extra="forbid")

    to: StrictStr = Field(min_length=3, max_length=255)
    subject: StrictStr = Field(min_length=1, max_length=200)
    body: StrictStr = Field(min_length=1, max_length=10_000)
    cc: Optional[StrictStr] = Field(default=None, max_length=255)

    @model_validator(mode="after")
    def check_subject_body(self):
        # example cross-field validation
        if "http" in self.subject.lower() and len(self.body) < 50:
            raise ValueError("Если в subject есть ссылка, body должен быть длиннее 50 символов")
        return self

@app.post("/emails")
async def send_email(payload: SendEmailIn, mailer: MailerClient = Depends(get_mailer)):
    try:
        result = await mailer.send(
            to=payload.to,
            subject=payload.subject,
            body=payload.body,
        )
        return result
    except RuntimeError as e:
        # если ресурс не готов — это скорее 503
        raise HTTPException(status_code=503, detail=str(e))

Что здесь достигается:

• Ресурс создаётся один раз в lifespan и закрывается при остановке.
• Зависимость просто выдаёт ссылку на ресурс — без повторной инициализации.
• Валидация запроса строгая: лишние поля запрещены, типы строгие, ограничения размеров есть.
• Бизнес-валидация выполняется в модели до вызова mailer.send.

Типичные ошибки и как их избегать

1) Закрывать “lifespan-синглтон” в dependency

Если объект создан в lifespan, он должен закрываться только там. Dependency должна быть без yield (или без финализаторов), если вы не планируете request-scoped жизненный цикл.

2) Использовать sync-код внутри async lifespan без оговорок

Если вы поднимаете “тяжёлый” синхронный клиент в async lifespan, блокировка event loop возможна. В таких случаях:

• либо используйте асинхронные библиотеки,
• либо выносите в отдельный thread через run_in_threadpool (но это отдельная тема),
• либо оставляйте создание минимальным и выполняйте “тяжёлое” в фоне.

3) Пытаться “подправить типы” прямо в endpoint

Если вы обнаружили, что клиент прислал "123" вместо 123, и вы делаете int(payload.user_id) вручную — это симптом. Лучше определить контракт и использовать StrictInt (или наоборот, позволить преобразование и явно задокументировать). И главное — держать преобразования в модели, а не размазывать по handlers.

4) Непредсказуемые сообщения об ошибках

Если часть проверок — в модели, а часть — в endpoint с произвольными HTTPException(detail=...), клиент получает неодинаковый формат ошибок. По возможности:

• бизнес-правила размещайте в модели через валидаторы;
• для “технических” отказов используйте HTTPException, но держите структуру и сообщения стабильными.

5) Раздувать dependency сложной бизнес-логикой

Dependency должна предоставлять зависимости и (иногда) легковесную подготовку. Если dependency превращается в мини-сервис, сложность растёт, а тестирование эндпоинта становится мутным.

Как тестировать это без боли

Хорошая валидация и корректный lifespan проще всего подтверждаются тестами.

1. Тесты модели Pydantic