FastAPI로 실전 비동기 API 만들기

FastAPI는 def와 async def를 다르게 취급한다

FastAPI를 처음 쓰면 헷갈리는 지점이 있습니다. 엔드포인트를 def 로 짜도 되고 async def 로 짜도 되는데, 둘 다 "작동은" 합니다. 차이는 성능 특성입니다. def 로 선언한 엔드포인트는 FastAPI가 내부적으로 스레드 풀에서 실행합니다(Starlette가 run_in_threadpool 로 감쌉니다). 반면 async def 로 선언하면 이벤트 루프에서 직접 실행됩니다.

from fastapi import FastAPI

app = FastAPI()

@app.get("/sync-endpoint")
def sync_endpoint():
    # 스레드 풀에서 실행됨 — 이 안에서 time.sleep을 써도 다른 요청은 안 막힘
    return {"ok": True}

@app.get("/async-endpoint")
async def async_endpoint():
    # 이벤트 루프에서 직접 실행됨 — 여기서 blocking 호출을 하면 서버 전체가 멈춤
    return {"ok": True}

이게 왜 중요하냐면, async def 로 선언해놓고 그 안에서 동기 라이브러리(동기 DB 드라이버, requests, CPU 연산)를 그대로 호출하는 게 6장에서 본 "이벤트 루프 전체를 멈추는" 사고의 가장 흔한 실전 발생 지점이기 때문입니다. 아이러니하게도 def 로 그냥 뒀으면 스레드 풀이 알아서 격리해줬을 코드를, async def 로 "더 빠르게 만들려다가" 오히려 전체 서버를 막는 경우를 실무에서 자주 봅니다. 원칙은 단순합니다 — 함수 안의 모든 I/O가 진짜로 논블로킹(await 가능)일 때만 async def 를 씁니다. 확신이 없으면 def 로 두는 편이 안전합니다.

DB 커넥션 풀 — 요청마다 새로 열지 않는다

비동기 DB 드라이버(asyncpg, SQLAlchemy의 AsyncEngine)를 쓸 때 흔한 실수는 요청마다 커넥션을 새로 여는 것입니다. FastAPI의 lifespan 으로 앱 시작 시 커넥션 풀을 한 번만 만들고, 요청마다는 풀에서 커넥션을 빌려 쓰는 구조가 맞습니다.

from contextlib import asynccontextmanager
from fastapi import FastAPI
import asyncpg

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.pool = await asyncpg.create_pool(
        dsn="postgresql://user:pass@localhost/db",
        min_size=5,
        max_size=20,
    )
    yield
    await app.state.pool.close()

app = FastAPI(lifespan=lifespan)

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    async with app.state.pool.acquire() as conn:
        row = await conn.fetchrow("SELECT id, name FROM users WHERE id = $1", user_id)
    return dict(row) if row else {"error": "not found"}

min_size/max_size 는 실제 DB의 max_connections 설정과 맞물려 있습니다. 워커 프로세스를 여러 개(예: uvicorn --workers 4) 띄운다면, 프로세스마다 독립된 풀이 생기므로 max_size × 워커 수 가 DB가 버틸 수 있는 총 커넥션 수를 넘지 않는지 반드시 확인해야 합니다. 이건 로컬 개발 환경에서는 절대 드러나지 않고, 트래픽이 몰리는 운영 환경에서 too many connections 에러로 처음 발견되는 종류의 문제입니다.

의존성 주입에서의 async generator

FastAPI의 Depends 는 요청 단위로 리소스를 만들고 정리하는 데 async def 제너레이터를 씁니다. 세션 하나를 요청마다 빌리고 요청이 끝나면 반납하는 패턴이 대표적입니다.

from sqlalchemy.ext.asyncio import AsyncSession

async def get_session(request: Request):
    async with request.app.state.session_factory() as session:
        yield session

@app.get("/posts/{post_id}")
async def get_post(post_id: int, session: AsyncSession = Depends(get_session)):
    result = await session.get(Post, post_id)
    return result

yield 이전 코드는 요청 처리 전에, yield 이후 코드(여기서는 async with 블록을 빠져나가며 실행되는 세션 정리)는 응답이 만들어진 뒤에 실행됩니다. 예외가 나더라도 async with 덕분에 세션은 정리됩니다 — 다만 트랜잭션 커밋/롤백 시점을 직접 제어해야 한다면 try/except 를 이 함수 안에 명시적으로 넣는 편이 안전합니다. FastAPI가 알아서 커밋해주지는 않습니다.