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

В FastAPI «вроде бы просто написать роут» — и это правда. Но в реальной эксплуатации почти всё решают жизненный цикл приложения, корректная инициализация внешних ресурсов (БД, очереди, HTTP-клиенты), закрытие соединений и то, как зависимости организованы относительно контекста запроса. Ошибки здесь чаще всего выглядят не как «оно не работает», а как редкие утечки, зависания при деплое, гонки при старте/остановке, некорректное использование общих клиентов и нестабильное поведение под нагрузкой.

В этой статье разберём практический, «продакшн-ориентированный» подход к FastAPI: как связать роутинг с жизненным циклом, как правильно использовать startup/shutdown, как организовать зависимости (включая контекст запроса), и как избежать проблем гонок и некорректного закрытия ресурсов. В конце — чек‑лист, который можно взять как основу для ревью вашего проекта.

Как FastAPI исполняет приложение: что реально происходит

FastAPI работает поверх ASGI (обычно через Uvicorn). ASGI-совместимый сервер вызывает у приложения события жизненного цикла и передаёт события/запросы в асинхронный цикл.

Ключевой момент: жизненный цикл приложения — это не «один запуск на один процесс» в абстрактном смысле. В продакшене вы почти всегда имеете:

• несколько воркеров (например, --workers в gunicorn/uvicorn),
• перезапуски контейнеров,
• graceful shutdown,
• отмены задач и таймауты.

Поэтому любые «глобальные» ресурсы нужно создавать осознанно: как минимум — на уровне процесса, а также помнить, что «стартап» и «шатаун» могут происходить в разное время на разных воркерах.

События startup и shutdown: где создавать и закрывать ресурсы

Почему не стоит создавать клиенты прямо в модуле

Самая распространённая ошибка новичков — создать, например, клиент БД «на уровне импорта» (в верхнем уровне файла). Это приводит к проблемам:

• инициализация может произойти до корректного запуска event loop,
• в тестах/сборке модуль может импортироваться без ожидаемой конфигурации,
• перезапуски/воркеры создают лишние клиенты,
• закрытие может быть забыто или выполнено слишком поздно.

Вместо этого используйте события жизненного цикла или dependency injection.

Практический шаблон: хранение ресурсов в app.state

FastAPI предоставляет app.state — удобное место для хранения «процессных» ресурсов. Принцип: создаём в startup, кладём в app.state, используем на роутерах/в зависимостях, закрываем в shutdown.

Ниже — пример с «условной» БД (асинхронный клиент) и HTTP-клиентом.

from fastapi import FastAPI
import httpx
from contextlib import AsyncExitStack

app = FastAPI()
# На практике лучше типизировать, но опустим для краткости.

@app.on_event("startup")
async def on_startup() -> None:
    # AsyncExitStack удобен тем, что вы централизованно
    # регистрируете "закрывалки" любых ресурсов.
    stack = AsyncExitStack()

    # HTTP клиент:
    http_client = httpx.AsyncClient(timeout=10.0)

    # Регистрация закрытия:
    await stack.enter_async_context(http_client)

    # Здесь же можно создать пул БД, подключение к брокеру очередей и т.п.
    # db = ...
    # await stack.enter_async_context(db_context_manager)

    app.state.exit_stack = stack
    app.state.http = http_client

@app.on_event("shutdown")
async def on_shutdown() -> None:
    stack = getattr(app.state, "exit_stack", None)
    if stack is not None:
        # Закроет всё, что было зарегистрировано в exit_stack.
        await stack.aclose()

Что важно

• startup/shutdown живут в рамках конкретного воркера процесса.
• AsyncExitStack снижает шанс «забыть закрыть» и делает порядок закрытия предсказуемым.
• shutdown должно быть максимально идемпотентным: сервер может инициировать закрытие несколько раз (или вы можете столкнуться с ошибками в логике).

Но есть нюанс: приоритет зависимостей и контекст запроса

Иногда ресурс нужно создавать не один раз на процесс, а «на запрос» (например, транзакцию или сессию). В этом случае лучше:

• использовать зависимости (Depends) с yield,
• создавать контекст транзакции на каждый запрос,
• закрывать/откатывать на выходе из зависимости.

Общая схема: dependency как контекстный менеджер

Предположим, у нас есть async_sessionmaker для SQLAlchemy Async или похожий фабрикат. Суть всегда одинаковая: зависимость на входе берёт «общий ресурс» (например, engine/pool), открывает сессию на запрос, а затем гарантированно закрывает её.

Пример (упрощённый; под вашу ORM подставите реальные вызовы):

from fastapi import Depends, Request
from typing import AsyncIterator

async def get_db_session(request: Request) -> AsyncIterator["AsyncSession"]:
    # engine/pool хранится на уровне app.state, создано в startup
    session_factory = request.app.state.session_factory

    session = session_factory()
    try:
        yield session
        # commit/rollback зависит от ваших правил транзакций
        # часто коммит делают на уровне сервиса после успешных операций
    except Exception:
        await session.rollback()
        raise
    finally:
        await session.close()

А в роутере:

from fastapi import APIRouter, Depends

router = APIRouter()

@router.get("/items/{item_id}")
async def read_item(item_id: int, session=Depends(get_db_session)):
    # Здесь session гарантированно жив до конца обработки запроса
    ...

Почему это правильно

• Конкретная сессия не разделяется между запросами (что снижает риск гонок внутри сессии).
• Закрытие/rollback выполняется даже при исключениях.
• Контекст запроса прозрачен: зависимости формируют правильную «обвязку».

Инициализация клиентов: БД, очереди, HTTP — однотипные ошибки

1) HTTP-клиент и connection pooling

HTTP-клиент обычно дорог по созданию и выигрывает от reuse (connection pooling, DNS caching, keep-alive). Поэтому:

• создавайте httpx.AsyncClient в startup,
• используйте его из зависимостей/роутеров.

Пример зависимости на получение клиента:

from fastapi import Request

def get_http_client(request: Request):
    return request.app.state.http

И использование:

from fastapi import Depends

@router.get("/proxy")
async def proxy(url: str, http=Depends(get_http_client)):
    resp = await http.get(url)
    resp.raise_for_status()
    return resp.json()

Подводный камень: общий клиент и отмены

Если запрос отменён (например, клиент закрыл соединение), то исключение CancelledError должно корректно всплыть и не оставить «подвисшие» операции. Правильно настроенные await/raise в httpx обычно справляются, но вы должны не подавлять отмены слишком агрессивно в middleware/обработчиках.

2) Очереди/клиенты брокера

Брокеры (RabbitMQ/Kafka/Redis Streams) имеют «соединение» и «каналы/потребителей». Типичная ошибка — создать consumer на каждый запрос. Потребитель обычно должен жить как задача фонового процесса или worker.

В FastAPI жизненный цикл позволяет:

• создать подключение в startup,
• стартовать фоновые задачи (например, consumer loop),
• остановить их в shutdown.

Это отдельная тема, но базовая заготовка такая:

import asyncio
from fastapi import FastAPI

app = FastAPI()

@app.on_event("startup")
async def startup():
    app.state.stop_event = asyncio.Event()
    app.state.bg_task = asyncio.create_task(background_worker(app.state.stop_event))

async def background_worker(stop_event: asyncio.Event):
    while not stop_event.is_set():
        # poll/broker consume
        await asyncio.sleep(0.1)

@app.on_event("shutdown")
async def shutdown():
    app.state.stop_event.set()
    task = app.state.bg_task
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        pass

Важно: при корректном graceful shutdown не только выставить stop_event, но и понимать, когда отменять задачи, чтобы не зависнуть.

3) БД и пул соединений

Пул (engine/connection pool) создаётся на процесс. Сессии — на запрос. Если наоборот — вы почти гарантированно получите проблемы:

• при запросах к одной сессии,
• взаимные блокировки,
• утечки соединений.

Роутинг + жизненный цикл: как «привязать» зависимости к конкретному app

Иногда код разносится по модулям: роуты отдельно, зависимости отдельно, клиенты отдельно. Ошибка архитектуры — создать зависимости так, чтобы они опирались на глобальные переменные (global engine, global http_client). Это усложняет:

• тестирование,
• запуск нескольких app (например, при ASGI mount),
• корректность в многоворкерной среде.

Правильный подход:

• клиенты держите в app.state,
• зависимости принимайте через Request (или напрямую request: Request в зависимостях),
• используйте request.app.state.

Так вы гарантируете, что зависимости читают данные именно из текущего приложения (а не «какого-то глобального»).

Избегаем гонок при старте/остановке: реальная проблема

Что может пойти не так

Представьте сценарий:

1. Сервер запущен, начинается startup.
2. В это время (особенно при некоторых настройках прокси/нагрузки) уже могут прийти запросы.
3. startup ещё не завершился, app.state.http или app.state.session_factory ещё не инициализированы.
4. Запрос падает с AttributeError или (хуже) начинает использовать частично созданный ресурс.

И обратная сторона:

1. Начинается shutdown.
2. В процессе закрытия ресурсов приходят запросы или они ещё не завершились.
3. Ресурсы уже закрыты, а обработчик запросов ещё их использует.
4. Получаете ошибки уровня «соединение закрыто».

ASGI-серверы обычно стараются делать graceful shutdown корректно, но на практике вы должны защищаться в приложении.

Защита №1: отложенная доступность ресурсов

Самый простой путь — завести флаг «готовности» и проверять в зависимостях.

from fastapi import FastAPI, Request, HTTPException
from typing import AsyncIterator
import asyncio
import httpx

app = FastAPI()
app.state.ready = False

@app.on_event("startup")
async def startup():
    app.state.http = httpx.AsyncClient(timeout=10.0)
    app.state.ready = True

@app.on_event("shutdown")
async def shutdown():
    app.state.ready = False
    await app.state.http.aclose()

def require_ready(request: Request):
    if not getattr(request.app.state, "ready", False):
        # 503 — сервис недоступен, клиенту логично повторить позже
        raise HTTPException(status_code=503, detail="App is not ready yet")

def get_http_client(request: Request):
    require_ready(request)
    return request.app.state.http

Да, это добавляет проверку в каждой зависимости, зато убирает редкие гонки.

Защита №2: гарантировать закрытие после завершения обработчиков

В корректной конфигурации ASGI-запросы должны завершиться до финального закрытия приложения. Но если вы запускаете фоновые задачи, отменяете их принудительно или закрываете ресурсы раньше, важно:

• либо дождаться завершения задач,
• либо прекращать приём новых запросов (на уровне сервера),
• либо делать «soft stop» фоновых задач.

С точки зрения кода — храните задачи в app.state и корректно управляйте их жизненным циклом в shutdown.

Защита №3: идемпотентное закрытие и аккуратная обработка исключений

В shutdown вы можете поймать ошибку закрытия одного ресурса. Не стоит допускать «цепочку отказов», когда ошибка в закрытии HTTP-клиента мешает закрыть БД.

Пример устойчивого shutdown:

@app.on_event("shutdown")
async def shutdown():
    app.state.ready = False
    errors = []

    http = getattr(app.state, "http", None)
    if http is not None:
        try:
            await http.aclose()
        except Exception as e:
            errors.append(e)

    # аналогично закрываем другие ресурсы...

    if errors:
        # Логируйте, но не обязательно падайте.
        # Лучше собрать в список и отправить в logging.
        # raise errors[0]  # часто не нужно
        pass

Dependency context и транзакции: где часто ломаются

Частая ошибка: делать commit в зависимости без учёта бизнес-логики

Если зависимость сама коммитит транзакцию «всегда после yield», вы не контролируете ситуации:

• когда бизнес-логика должна решить, коммитить или откатывать,
• когда некоторые исключения не требуют rollback,
• когда вы хотите использовать nested transactions.

Более гибко:

• коммит делайте в сервисе после успешного выполнения,
• зависимость отвечает за rollback при «неуспехе» и за закрытие.

Правильный паттерн: rollback/close, коммит — отдельно

from fastapi import Request
from typing import AsyncIterator

async def get_session(request: Request) -> AsyncIterator["AsyncSession"]:
    factory = request.app.state.session_factory
    session = factory()
    try:
        yield session
    except Exception:
        # rollback только при исключении
        await session.rollback()
        raise
    finally:
        await session.close()

А в роуте/сервисе:

@router.post("/orders")
async def create_order(payload: dict, session=Depends(get_session)):
    order = ...  # create entities
    session.add(order)
    await session.commit()
    await session.refresh(order)
    return order

Избегаем «перепутали уровни»

• startup: создать общие long-lived ресурсы (пулы, клиенты, фабрики).
• Зависимости на запрос: создать короткоживущие контексты (сессии, транзакции).
• Бизнес-уровень: принимать решение про commit/rollback.

Стартап как этап подготовки: миграции, валидация конфигурации, прогрев

В startup часто хотят:

• прогреть кеши,
• проверить доступность БД,
• выполнить миграции,
• поднять consumers.

Но есть практическое правило: startup не должен превращаться в «долгую операцию на десятки секунд». Если вы делаете миграции при старте — сервис может не подняться или будет нестабильно перезапускаться.

Здравый компромисс:

• В startup — проверить базовую доступность ресурсов и инициализировать клиенты.
• Миграции — обычно отдельным job/этапом деплоя.
• Прогрев — опционально и недетерминированно (или с контролем таймаутов).

Если всё же делаете check-подготовку, используйте таймауты и логирование.

Типовые ошибки в проектах на FastAPI

1. Глобальные клиенты без контроля жизненного цикла
   Почти гарантированно ломается при тестах, при нескольких инстансах и при корректном shutdown.

2. Создание сессии/транзакции на модуль
   Сессия не предназначена для переиспользования между запросами.

3. Отсутствие rollback в зависимостях
   После исключений вы можете оставлять транзакцию открытой/в неправильном состоянии.

4. Подавление отмены задач (CancelledError)
   Это приводит к зависаниям и «залипшим» запросам при отмене.

5. Нет защиты от гонок при старте
   В редких условиях (особенно при нестандартных прокси и таймингах) запрос может прийти до готовности ресурсов.

6. Неидемпотентный shutdown
   Повторный вызов (или частичное создание ресурсов) приводит к новым ошибкам во время остановки.

7. Фоновые задачи без корректной остановки
   При shutdown контейнер может быть убит, оставив незаконченную работу и неконсистентность.

Чек-лист для продакшена: что проверить перед релизом

Жизненный цикл

• Ресурсы long-lived (HTTP-клиент, пул БД, фабрики, продюсеры/коннекторы) создаются в startup.
• Ресурсы закрываются в shutdown (желательно централизованно, например через AsyncExitStack).
• shutdown идемпотентен: повторный вызов не ломает приложение.

Зависимости

• На запрос создаются короткоживущие контексты (сессия/транзакция), а не общий объект.
• В dependency используется try/except/finally с rollback по исключению и close всегда.
• Зависимости берут ресурсы из request.app.state, а не из глобальных переменных.

Гонки и готовность

• Есть защита от обработки запросов до завершения startup (например, флаг ready и 503).
• В shutdown выставляется готовность False или производится иной механизм остановки новых действий.
• Фоновые задачи корректно останавли