FastAPI의 Connection Lifecycle

~14 min · fastapi, lifespan, aiosqlite

Level 0Scout

0 XP0/80 lessons0/10 achievements

0/120 XP to next level120 XP to go0% complete

한 번 열고 한 번 닫아

FastAPI는 앱이 뜰 때와 내려갈 때 도는 lifespan context manager를 줘. 정석은 이래. 거기서 DB connection을 열어 app.state에 붙여두고, dependency injection으로 route에 꽂아주는 거야.

피해야 할 함정이 둘 있어.

request마다 새 connection을 여는 것. 파일 핸들을 낭비하고 메모리에 쌓아둔 캐시도 못 써.
module을 import하는 시점에 여는 것. 그때는 event loop가 아직 없는데 aiosqlite는 loop가 있어야 해.

Self-reference: 피파 backend/main.py가 정확히 이 lifespan 패턴이야. 뜰 때 SQLite store를 한 번 열고, migration을 돌리고, app.state.store에 붙이고, 내려갈 때 깔끔하게 닫아. route는 Depends로 store를 받아 쓰고.

Code

FastAPI + aiosqlite lifespan·python

from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends, Request
import aiosqlite

@asynccontextmanager
async def lifespan(app: FastAPI):
    conn = await aiosqlite.connect('myapp.db')
    conn.row_factory = aiosqlite.Row
    await conn.execute('PRAGMA journal_mode = WAL')
    await conn.execute('PRAGMA foreign_keys = ON')
    await conn.execute('PRAGMA busy_timeout = 5000')
    app.state.db = conn
    try:
        yield
    finally:
        await conn.close()

app = FastAPI(lifespan=lifespan)

async def db(request: Request) -> aiosqlite.Connection:
    return request.app.state.db

@app.get('/notes')
async def list_notes(conn: aiosqlite.Connection = Depends(db)):
    async with conn.execute('SELECT id, body FROM notes ORDER BY id DESC LIMIT 50') as cur:
        return [dict(row) async for row in cur]

External links

Exercise

lifespan 패턴으로 aiosqlite를 물린 endpoint 하나짜리 작은 FastAPI 앱을 만들어봐. uvicorn으로 띄우고 hey나 ab로 동시 요청 1000개를 때려. 지연이 1밀리초 아래로 유지되고 'database is locked' 에러가 안 나는지 확인해.

Progress

Progress is local-only — sign in to sync across devices.

← PreviousAsync cursor, iteration, transaction Next →Single Connection vs Pool

이 페이지에서 버그를 발견하셨거나 피드백이 있으세요?문제 신고

🔔 답글 알림 (로그인 필요)

로그인 — 댓글을 남기려면 로그인해 주세요.

아직 댓글이 없어요. 첫 댓글을 남겨보세요.