Builder.shin

FastAPI에서 SQLAlchemy async 세션 관리 개선

배경

회사 프로젝트에서 FastAPI를 사용하는데, SQLAlchemy를 동기 방식으로 사용하다 보니 비동기 엔드포인트에서 블로킹이 발생했다. SQLAlchemy 1.4가 asyncio를 정식 지원하면서 마이그레이션을 진행했다.

기존 방식의 문제

# 동기 방식
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/users/{user_id}")
def get_user(user_id: int, db: Session = Depends(get_db)):
    return db.query(User).filter(User.id == user_id).first()

비동기 엔드포인트에서 동기 DB 호출이 이벤트 루프를 블로킹했다. run_in_executor로 우회할 수도 있지만 근본적인 해결은 아니었다.

async 세션 적용

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,
)

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

async def get_db():
    async with AsyncSessionLocal() 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))
    return result.scalars().first()

주의사항

  1. 드라이버 변경: psycopg2 대신 asyncpg 사용
  2. 쿼리 방식: .query() 대신 select() 사용
  3. 결과 처리: .execute() 후 .scalars() 또는 .fetchall() 필요

성능 개선

부하 테스트 결과 동시 요청 100개 기준으로 응답 시간이 평균 40% 감소했다. DB 연결 대기 시간이 사라지면서 전체 처리량도 개선됐다.

마이그레이션 팁

  • 모델 정의는 그대로 유지 가능
  • relationship lazy loading은 selectinload 명시 필요
  • 트랜잭션 처리는 async with session.begin() 사용

기존 코드베이스가 크지 않아서 하루 정도 작업으로 마이그레이션을 완료했다. SQLAlchemy 1.4의 타입 힌팅도 개선되어 코드 안정성도 함께 올라갔다.

FastAPI에서 SQLAlchemy async 세션 관리 개선