【Python】AsyncのLifespanとは？

FastAPI や Starlette アプリケーションでは、アプリケーションの起動時・終了時に非同期処理を実行する仕組みとして lifespan という概念があります。

from fastapi import FastAPI
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    print("🌱 Startup: connecting to resources...")
    yield  # この間にリクエスト処理が行われる
    print("🍂 Shutdown: cleaning up...")

app = FastAPI(lifespan=lifespan)

@app.get("/")
async def hello():
    return {"message": "Hello"}

同時接続数

lifespan はアプリケーションの起動時にカウンターや状態を初期化し、WebSocket や HTTP ハンドラの中でそのカウンターを インクリメント／デクリメント すれば、アプリ全体で接続数をトラッキングできます。

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.connection_count = 0
    app.state.conn_lock = asyncio.Lock()
    yield

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()

    async with app.state.conn_lock:
        app.state.connection_count += 1
        print(f"🔌 Connected: {app.state.connection_count}")

    try:
        while True:
            await websocket.receive_text()
    except Exception:
        pass
    finally:
        async with app.state.conn_lock:
            app.state.connection_count -= 1
            print(f"❌ Disconnected: {app.state.connection_count}")