Lifespan события в FastAPI

Lifespan — управление запуском и остановкой приложения (инициализация БД, ML-модели, пулы соединений).

Современный способ (FastAPI 0.95+)

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: выполняется при запуске
    print("Запуск приложения...")
    await init_db()
    yield
    # Shutdown: выполняется при остановке
    print("Остановка приложения...")
    await close_db()

app = FastAPI(lifespan=lifespan)

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

Инициализация БД

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Создать таблицы при старте
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    # Закрыть пул соединений
    await engine.dispose()

app = FastAPI(lifespan=lifespan)

Загрузка ML-модели

ml_model = {}

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Загрузить модель один раз при старте
    ml_model["classifier"] = load_model("model.pkl")
    yield
    # Очистить
    ml_model.clear()

app = FastAPI(lifespan=lifespan)

@app.post("/predict/")
def predict(data: PredictRequest):
    result = ml_model["classifier"].predict(data.features)
    return {"result": result}

Redis и HTTP-клиент

import httpx
import redis.asyncio as redis

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.redis = await redis.from_url("redis://localhost")
    app.state.http_client = httpx.AsyncClient()
    yield
    await app.state.redis.close()
    await app.state.http_client.aclose()

Доступ к состоянию в endpoints

@app.get("/cache/{key}")
async def get_cache(key: str, request: Request):
    value = await request.app.state.redis.get(key)
    return {"value": value}

Устаревший способ (startup/shutdown события)

# Старый способ — работает, но не рекомендуется
@app.on_event("startup")
async def startup():
    await init_db()

@app.on_event("shutdown")
async def shutdown():
    await close_db()