Builder.shin

FastAPI에서 비동기 DB 쿼리 최적화 경험

문제 상황

회사 API 서버를 Flask에서 FastAPI로 마이그레이션하는 작업을 진행했다. 비동기 처리의 이점을 살리기 위해 SQLAlchemy 1.4의 async 기능과 asyncpg를 도입했는데, 특정 엔드포인트에서 응답 시간이 예상보다 길게 나왔다.

# 문제가 있던 코드
@app.get("/users/{user_id}/posts")
async def get_user_posts(user_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(Post).where(Post.user_id == user_id))
    posts = result.scalars().all()
    
    # N+1 발생 지점
    for post in posts:
        comments = await post.awaitable_attrs.comments
    
    return posts

원인 분석

전형적인 N+1 쿼리 문제였다. 각 post마다 comments를 개별적으로 조회하면서 DB 왕복이 반복됐다. 비동기라고 해서 자동으로 최적화되는 게 아니었다.

해결 방법

selectinload를 사용해서 한 번에 로딩하도록 수정했다.

from sqlalchemy.orm import selectinload

@app.get("/users/{user_id}/posts")
async def get_user_posts(user_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(
        select(Post)
        .options(selectinload(Post.comments))
        .where(Post.user_id == user_id)
    )
    posts = result.scalars().all()
    return posts

응답 시간이 평균 800ms에서 120ms로 개선됐다. SQLAlchemy의 비동기 API는 아직 문서가 부족한 편이라 시행착오가 있었지만, 기본적인 ORM 최적화 원칙은 동일하게 적용된다는 걸 다시 확인했다.

추가 고려사항

  • joinedload는 LEFT OUTER JOIN을 사용하므로 일대다 관계에서는 중복 결과가 나올 수 있다
  • selectinload는 별도 쿼리를 사용하지만 결과가 예측 가능하다
  • 프로파일링 없이 premature optimization은 하지 않기로 했다
FastAPI에서 비동기 DB 쿼리 최적화 경험