Перейти к содержанию

Управление жизненным циклом приложения

Модуль lifespan предоставляет функционал для управления жизненным циклом FastAPI приложения, обеспечивая корректную инициализацию и освобождение ресурсов.

Особенности

  • Асинхронное управление жизненным циклом
  • Безопасная инициализация и освобождение ресурсов
  • Интеграция с системой бутстраппинга
  • Обработка ошибок при старте и завершении

Документация API

lifespan

Асинхронный контекстный менеджер для управления жизненным циклом приложения.

Source code in src/infrastructure/lifespan.py 
13
14
15
16
17
18
19
20
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    try:
        await bootstrap()
        yield
    finally:
        engine = get_engine()
        await engine.dispose()

Примеры использования

Базовое использование с FastAPI

from fastapi import FastAPI
from src.infrastructure.lifespan import lifespan

# Создание приложения с кастомным жизненным циклом
app = FastAPI(lifespan=lifespan)

# Определение маршрутов...
@app.get("/")
async def root():
    return {"message": "Приложение работает"}

Кастомизация жизненного цикла

from contextlib import asynccontextmanager
from fastapi import FastAPI
from src.infrastructure.lifespan import lifespan as base_lifespan

@asynccontextmanager
async def custom_lifespan(app: FastAPI):
    # Дополнительная инициализация перед запуском
    print("Выполняется дополнительная инициализация...")

    # Использование базового жизненного цикла
    async with base_lifespan(app) as manager:
        # Код выполняется после инициализации приложения
        print("Приложение готово к работе")
        try:
            yield manager
        finally:
            # Дополнительная очистка при завершении
            print("Выполняется дополнительная очистка...")

app = FastAPI(lifespan=custom_lifespan)

Рекомендации по использованию

  1. Инициализация ресурсов
  2. Используйте bootstrap() для централизованной инициализации
  3. Инициализируйте только необходимые ресурсы

  4. Обрабатывайте ошибки при инициализации

  5. Освобождение ресурсов

  6. Гарантируйте освобождение всех ресурсов в блоке finally
  7. Используйте асинхронные менеджеры контекста

  8. Обрабатывайте ошибки при освобождении ресурсов

  9. Логирование

  10. Логируйте ключевые этапы жизненного цикла
  11. Включайте в логи информацию об ошибках
  12. Используйте разные уровни логирования

Интеграция

Модуль интегрируется с:

  • FastAPI приложениями
  • Системой бутстраппинга
  • Движком базы данных
  • Системой логирования

Ограничения

  • Требуется Python 3.8+
  • Зависит от FastAPI с поддержкой lifespan
  • Предполагает использование асинхронного кода

Дополнительные возможности

Добавление хуков жизненного цикла

from fastapi import FastAPI
from src.infrastructure.lifespan import lifespan as base_lifespan

app = FastAPI()

# Добавление обработчиков событий жизненного цикла
@app.on_event("startup")
async def startup_event():
    print("Дополнительная инициализация при запуске")

@app.on_event("shutdown")
async def shutdown_event():
    print("Дополнительная очистка при завершении")

# Использование базового жизненного цикла
app.router.lifespan_context = base_lifespan

Обработка ошибок при инициализации

from fastapi import FastAPI, HTTPException
from src.infrastructure.lifespan import lifespan as base_lifespan

@asynccontextmanager
async def safe_lifespan(app: FastAPI):
    try:
        async with base_lifespan(app):
            yield
    except Exception as e:
        # Логирование критической ошибки
        print(f"Критическая ошибка при инициализации: {e}")

        # Генерация понятной ошибки для пользователя
        @app.get("/")
        async def error_route():
            raise HTTPException(
                status_code=503,
                detail="Сервис временно недоступен из-за ошибки инициализации"
            )

        # Позволяет приложению запуститься в деградированном режиме
        yield

app = FastAPI(lifespan=safe_lifespan)