FastAPI Lifespan으로 앱의 시작과 끝을 관리하는 법

# ❌ 안티패턴: 매 요청마다 모델 로딩
@app.post("/predict")
async def predict(data: PredictRequest):
    model = load_model("model_v3.pkl")  # 매번 수 초 소요
    result = model.predict(data.features)
    return {"prediction": result}

from fastapi import FastAPI

app = FastAPI()

db_connection = None

@app.on_event("startup")
async def startup():
    global db_connection
    db_connection = await create_db_pool("postgresql://localhost/mydb")

@app.on_event("shutdown")
async def shutdown():
    if db_connection:
        await db_connection.close()

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # --- Startup ---
    model = load_model("model_v3.pkl")
    print(f"Model loaded: {model}")

    yield  # 이 시점부터 요청을 받기 시작

    # --- Shutdown ---
    del model
    print("Model unloaded")

app = FastAPI(lifespan=lifespan)

# ✅ After: 하나의 함수에서 시작과 종료를 관리
@asynccontextmanager
async def lifespan(app: FastAPI):
    db_connection = await create_db_pool("postgresql://localhost/mydb")
    app.state.db = db_connection

    yield

    await db_connection.close()

@asynccontextmanager
async def lifespan(app: FastAPI):
    db = await create_db_pool("postgresql://localhost/mydb")
    app.state.db = db

    redis = None
    try:
        redis = await create_redis_client("redis://localhost")
        app.state.redis = redis
        yield
    finally:
        if redis:
            await redis.close()
        await db.close()

from contextlib import asynccontextmanager
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    engine = create_async_engine(
        "postgresql+asyncpg://user:pass@localhost/mydb",
        pool_size=20,
        max_overflow=10,
    )
    session_factory = async_sessionmaker(engine, expire_on_commit=False)

    app.state.db_engine = engine
    app.state.db_session_factory = session_factory

    yield

    await engine.dispose()

import httpx

@asynccontextmanager
async def lifespan(app: FastAPI):
    http_client = httpx.AsyncClient(
        timeout=httpx.Timeout(10.0),
        limits=httpx.Limits(max_connections=100),
    )
    app.state.http_client = http_client

    yield

    await http_client.aclose()

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 1. 설정 로딩
    settings = Settings()

    # 2. DB 커넥션 풀
    engine = create_async_engine(settings.database_url)
    app.state.db_engine = engine

    # 3. HTTP 클라이언트
    http_client = httpx.AsyncClient(timeout=10.0)
    app.state.http_client = http_client

    try:
        yield
    finally:
        # 정리는 초기화의 역순
        await http_client.aclose()
        await engine.dispose()

import time
import logging

logger = logging.getLogger(__name__)

class ModelManager:
    def __init__(self):
        self.models: dict[str, object] = {}
        self.is_loading = False

    @property
    def is_ready(self) -> bool:
        return bool(self.models) and not self.is_loading

    def load_all_models(self) -> None:
        """모든 모델을 로드한다. 새 dict를 완성한 후 한 번에 교체."""
        self.is_loading = True
        start = time.time()

        try:
            new_models = {}
            for name, path in MODEL_REGISTRY.items():
                logger.info("Loading %s from %s", name, path)
                new_models[name] = load_model(path)

            # 새 dict를 통째로 교체 (읽는 쪽은 참조를 한 번만 집어감)
            self.models = new_models
            logger.info("All models loaded in %.2fs", time.time() - start)
        finally:
            self.is_loading = False

    def get_model(self, name: str):
        model = self.models.get(name)
        if model is None:
            raise RuntimeError(f"'{name}' model not loaded")
        return model

@asynccontextmanager
async def lifespan(app: FastAPI):
    model_manager = ModelManager()
    model_manager.load_all_models()

    app.state.model_manager = model_manager
    logger.info("Startup complete. Models ready: %s", model_manager.is_ready)

    yield

class ModelManager:
    # ... (위의 코드에 이어서)

    def reload_models(self) -> dict:
        """서버 재시작 없이 모델 교체. 새 모델을 완성한 후 한 번에 교체."""
        if self.is_loading:
            raise RuntimeError("Model loading already in progress")

        start = time.time()
        self.load_all_models()

        return {
            "status": "success",
            "elapsed_seconds": round(time.time() - start, 2),
            "reloaded_models": list(self.models.keys()),
        }

from fastapi import APIRouter, Request

router = APIRouter()

@router.post("/admin/reload-models")
async def reload_models(request: Request):
    model_manager = request.app.state.model_manager
    result = model_manager.reload_models()
    return result

@router.get("/health")
async def health(request: Request):
    model_manager = request.app.state.model_manager
    return {
        "status": "ok",
        "models": list(model_manager.models.keys()),
    }

from fastapi import FastAPI, Request, Depends

async def get_db_session(request: Request):
    session_factory = request.app.state.db_session_factory
    async with session_factory() as session:
        yield session

@app.get("/users")
async def list_users(session = Depends(get_db_session)):
    result = await session.execute(select(User))
    return result.scalars().all()

from dependency_injector import containers, providers

class Container(containers.DeclarativeContainer):
    # lifespan에서 실제 인스턴스로 교체될 placeholder
    model_manager = providers.Object(None)

    settings = providers.Singleton(get_settings)

    scoring_service = providers.Singleton(
        ScoringService,
        model_manager=model_manager,  # placeholder 주입
    )

@asynccontextmanager
async def lifespan(app: FastAPI):
    container: Container = app.container

    model_manager = ModelManager()
    model_manager.load_all_models()

    # placeholder를 실제 인스턴스로 교체
    container.model_manager.override(model_manager)
    app.state.model_manager = model_manager

    yield

    # 리소스 정리 (Resource provider가 있을 때만 awaitable을 돌려줍니다)
    container.shutdown_resources()

import logging

logger = logging.getLogger(__name__)

def create_optional_connection(url: str, name: str):
    """선택 리소스 연결. 실패 시 None 반환."""
    try:
        return create_connection(url)
    except Exception:
        logger.exception("%s 연결 실패, 해당 기능 비활성화", name)
        return None

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 필수: 실패하면 crash (yield에 도달 못함)
    model_manager = ModelManager()
    model_manager.load_all_models()
    app.state.model_manager = model_manager

    # 선택: 실패해도 None으로 계속 진행
    cms_client = create_optional_connection(settings.cms_url, "CMS MySQL")
    app.state.cms_client = cms_client

    try:
        yield
    finally:
        if cms_client:
            cms_client.close()

app/
├── main.py              # create_app() + 라우터 등록
├── lifespan.py          # lifespan 함수
├── container.py         # DI Container
├── settings.py          # BaseSettings
├── scoring/
│   ├── model_manager.py # ML 모델 관리
│   ├── service.py
│   └── controller.py
└── health/
    └── controller.py    # /health, /admin/reload-models

# main.py
from fastapi import FastAPI
from app.lifespan import lifespan
from app.container import Container

def create_app() -> FastAPI:
    container = Container()
    app = FastAPI(lifespan=lifespan)
    app.container = container
    app.include_router(health_router)
    app.include_router(scoring_router)
    return app

app = create_app()

# ❌ 이러면 lifespan이 실행되지 않음
client = TestClient(app)
response = client.get("/health")

# ✅ with 문으로 감싸야 lifespan이 실행됨
with TestClient(app) as client:
    response = client.get("/health")
    assert response.status_code == 200

import pytest
from fastapi.testclient import TestClient
from app.main import app

@pytest.fixture
def client():
    with TestClient(app) as c:
        yield c

def test_health(client):
    response = client.get("/health")
    assert response.status_code == 200
    assert "models" in response.json()

def test_predict(client):
    response = client.post("/predict", json={"features": [1.0, 2.0]})
    assert response.status_code == 200