Как управлять жизненным циклом подключения через lifespan FastAPI

Управление жизненным циклом подключения через `lifespan` в FastAPI позволяет автоматически управлять зависимостями, которые должны быть запущены при старте приложения и остановлены при его завершении. Это стандартный паттерн для работы с базами данных, очередями сообщений, WebSocket-соединениями или внешними API. Механизм реализует контекстный менеджер, который перехватывает события `startup` и `shutdown`. При запуске сервера вызывается метод `__aenter__`, при остановке — `__aexit__`. Если приложение падает с ошибкой, блок `finally` в методе `__aexit__` всё равно выполнится, что гарантирует освобождение ресурсов.

Управление жизненным циклом подключения через `lifespan` в FastAPI: базовые принципы и архитектура

Модуль `lifespan` предоставляет два декоратора: `@app.on_event('startup')` и `@app.on_event('shutdown')`. Они вызываются соответственно при запуске и остановке приложения. Эти события срабатывают один раз на весь процесс, а не для каждого запроса.

Архитектурно это реализуется через класс `LifeSpanContextManager`. При создании экземпляра приложения он регистрирует обработчики событий в объекте `EventHandlers`. Когда сервер запускается (например, через `uvicorn`), вызывается метод `__aenter__`, который выполняет код из декоратора `startup`. При получении сигнала остановки (SIGINT, SIGTERM или завершение работы процесса) вызывается `__aexit__`, выполняющий код из `shutdown`.

Важно понимать разницу между жизненным циклом приложения и жизненным циклом запроса. Зависимости в `lifespan` живут столько, сколько живёт процесс сервера. Если приложение перезапускается (например, при обновлении кода), старые зависимости должны быть корректно закрыты перед созданием новых.

Настройка контекстного менеджера для автоматического старта и остановки зависимостей

Для настройки создайте файл `main.py` с приложением. Внутри определите зависимости, которые нужно инициализировать.

Пример кода для подключения к базе данных или очереди:

«`python
from contextlib import asynccontextmanager
import asyncio
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
# Старт
print("Запуск приложения…")
db_connection = await create_db_connection()
app.state.db = db_connection

yield # Точка, где приложение работает

# Остановка
print("Остановка приложения…")
if hasattr(app.state, 'db'):
await close_db_connection(app.state.db)

app = FastAPI(lifespan=lifespan)
«`

В этом примере `create_db_connection` и `close_db_connection` — это асинхронные функции. Ключевая часть — переменная `yield`. Она отделяет код запуска от кода остановки. Всё до `yield` выполняется при старте, всё после — при остановке.

При запуске сервера через `uvicorn main:app —reload`, вы увидите вывод сообщений о старте и остановке. Если приложение падает с ошибкой во время работы, блок остановки всё равно выполнится, так как он находится в блоке `finally` внутри реализации контекстного менеджера.

Реализация кастомных событий (start, stop) для интеграции с внешними сервисами

Стандартные события `startup` и `shutdown` подходят для большинства задач. Однако иногда требуется более гибкая логика, например, отправка уведомлений в Slack или запись логов в специфический формат.

Для этого можно использовать декораторы `@app.on_event('startup')` и `@app.on_event('shutdown')` внутри функции `lifespan` или создавать отдельные обработчики событий.

Пример интеграции с внешним сервисом:

«`python
@app.on_event('startup')
async def send_startup_notification():
await external_service.notify("Сервер запущен")

@app.on_event('shutdown')
async def send_shutdown_notification():
await external_service.notify("Сервер остановлен")
«`

Если внешние сервисы требуют асинхронного взаимодействия, убедитесь, что функции `notify` являются асинхронными. Если они синхронные, оберните их в `run_in_executor` или используйте библиотеку `anyio` для выполнения в отдельном потоке, чтобы не блокировать основной цикл событий FastAPI.

Обработка ошибок при остановке приложения и гарантированное освобождение ресурсов

Одна из главных задач `lifespan` — гарантированно освободить ресурсы даже при аварийном завершении. В реализации контекстного менеджера блок кода после `yield` выполняется в блоке `finally`. Это означает, что если во время остановки возникнет ошибка (например, таймаут при отключении БД), код освобождения ресурсов всё равно выполнится.

Однако важно обрабатывать исключения внутри блока остановки, чтобы не скрывать ошибки и не блокировать процесс завершения:

«`python
try:
await close_db_connection(app.state.db)
except Exception as e:
logger.error(f"Ошибка при остановке БД: {e}")
«`

Если критическая ошибка возникает во время остановки и приложение не может завершиться корректно, это может привести к зависанию процесса. В таких случаях рекомендуется использовать сигналы ОС (SIGTERM) для принудительной остановки с таймаутом.

Миграция логики из `lifespan` в асинхронные задачи или фоновые воркеры

Логика в `lifespan` выполняется один раз при старте/остановке. Если требуется выполнять периодические задачи (например, очистка кэша каждую минуту), `lifespan` не подходит. Для таких задач используйте фоновые воркеры через `BackgroundTasks` или планировщики вроде `APScheduler`.

Пример миграции:

«`python
from fastapi import BackgroundTasks

@app.post("/task")
async def create_task(task_data: dict, background_tasks: BackgroundTasks):
background_tasks.add_task(process_task, task_data)
«`

Если нужно запускать фоновые задачи при старте приложения, добавьте их в список `background_tasks` внутри функции `lifespan` или используйте отдельный цикл с `while True` и `await asyncio.sleep()`, но это требует осторожности, чтобы не блокировать процесс.

Тестирование корректности работы жизненного цикла в CI/CD пайплайнах

В CI/CD пайплайнах важно проверять, что приложение корректно запускается и останавливается. Для этого можно использовать скрипты, которые запускают сервер, отправляют запросы и затем принудительно останавливают процесс.

Пример теста на Python:

«`python
import subprocess
import time

def test_lifespan():
process = subprocess.Popen(['uvicorn', 'main:app'])
time.sleep(2) # Даем время на запуск

# Отправляем запрос
import requests
response = requests.get('http://localhost:8000/health')
assert response.status_code == 200

# Останавливаем процесс
process.terminate()
process.wait(timeout=5)
«`

Такие тесты можно добавить в пайплайн для автоматической проверки стабильности приложения перед деплоем.

Ограничения подхода и альтернативы для сложных сценариев (например, Kubernetes)

В Kubernetes приложение часто запускается в контейнере, который может быть остановлен сигналом `SIGTERM`. Если приложение не завершает работу за отведённое время (по умолчанию 30 секунд), kubelet принудительно убивает процесс. В этом случае код в `lifespan` может не успеть выполниться.

Для таких сценариев рекомендуется использовать библиотеки, которые обрабатывают сигналы ОС и дают больше времени на корректную остановку. Также можно настроить `preStop` хук в манифесте Kubernetes для выполнения дополнительных действий перед остановкой контейнера.

Альтернативой `lifespan` может быть использование `asyncio.run()` с кастомным контекстным менеджером, который явно обрабатывает сигналы.

Чек-лист по безопасности: очистка чувствительных данных при остановке сервиса

При остановке приложения важно очистить чувствительные данные из памяти, если они хранятся в зависимостях. Например, ключи API, токены или сессии пользователей.

Пример очистки:

«`python
@app.on_event('shutdown')
async def cleanup_sensitive_data():
if hasattr(app.state, 'api_key'):
app.state.api_key = None # Очистка из памяти
«`

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

Практические примеры и ошибки

Ошибка 1: Забыть закрыть соединение с БД. Решение: всегда добавляйте код закрытия соединения в блок после `yield`.

Ошибка 2: Использовать синхронные функции в асинхронном контексте. Решение: оберните их в `run_in_executor` или используйте асинхронные аналоги.

Ошибка 3: Не обрабатывать исключения при остановке. Решение: добавьте блок `try-except` вокруг кода остановки.

FAQ

Вопрос: Можно ли использовать `lifespan` для обработки каждого запроса? Ответ: Нет, `lifespan` вызывается один раз при старте/остановке. Для обработки запросов используйте зависимости в функциях маршрутов.

Вопрос: Что делать, если приложение не останавливается корректно? Ответ: Используйте сигналы ОС и настройте таймауты в конфигурации сервера.

Вопросы и ответы

Можно ли использовать `lifespan` для обработки каждого запроса?

Нет, `lifespan` вызывается один раз при старте и остановке приложения. Для обработки каждого запроса используйте зависимости в функциях маршрутов или фоновые задачи.

Что делать, если приложение не останавливается корректно?

Используйте сигналы ОС (SIGTERM) и настройте таймауты в конфигурации сервера. Также можно добавить `preStop` хук в манифесте Kubernetes для выполнения дополнительных действий перед остановкой.

Как гарантировать освобождение ресурсов при аварийном завершении?

Код после `yield` в контекстном менеджере выполняется в блоке `finally`, что гарантирует выполнение даже при ошибках. Обрабатывайте исключения внутри этого блока, чтобы не скрывать ошибки.

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Старт
    print("Запуск приложения...")
    db_connection = await create_db_connection()
    app.state.db = db_connection
    
    yield  # Точка, где приложение работает
    
    # Остановка
    print("Остановка приложения...")
    if hasattr(app.state, 'db'):
        await close_db_connection(app.state.db)