Builder.shin

FastAPI에서 비동기 DB 쿼리 처리하기

배경

회사 내부 API 서버를 Django에서 FastAPI로 전환하는 작업을 진행 중이다. FastAPI의 비동기 처리 장점을 살리기 위해 DB 레이어도 async/await 패턴으로 변경했다.

SQLAlchemy 비동기 설정

SQLAlchemy 1.4부터 asyncio를 공식 지원한다. asyncpg를 드라이버로 사용했다.

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/db",
    echo=True,
    future=True
)

async_session = sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False
)

의존성 주입

FastAPI의 Depends를 활용해 세션을 주입했다.

async def get_db():
    async with async_session() as session:
        yield session

@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(
        select(User).where(User.id == user_id)
    )
    user = result.scalar_one_or_none()
    return user

성능 비교

동시 요청 100개 기준으로 부하 테스트를 돌렸다.

  • Django (동기): 평균 응답시간 450ms
  • FastAPI (비동기): 평균 응답시간 180ms

I/O 대기 시간이 긴 쿼리에서 차이가 더 컸다.

주의사항

비동기 세션에서는 lazy loading이 동작하지 않는다. selectinload나 joinedload로 명시적 eager loading이 필요하다.

result = await db.execute(
    select(User).options(selectinload(User.posts))
)

기존 ORM 코드를 변환하면서 이 부분에서 에러가 많이 발생했다. 테스트 코드를 꼼꼼히 작성해두길 잘했다.

FastAPI에서 비동기 DB 쿼리 처리하기