FastAPI에서 SQLAlchemy 2.0 스타일로 마이그레이션하기

배경

회사 API 서버가 FastAPI + SQLAlchemy 1.4를 사용 중이었다. SQLAlchemy 2.0 릴리즈 준비 단계에서 1.4부터 지원하는 2.0 스타일 쿼리로 미리 전환하기로 결정했다.

주요 변경사항

1. Session 생성 방식

기존에는 sessionmaker를 직접 사용했지만, async 지원을 위해 AsyncSession으로 변경했다.

# Before
from sqlalchemy.orm import sessionmaker
SessionLocal = sessionmaker(bind=engine)

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

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

2. 쿼리 방식 변경

가장 큰 변화는 Query 객체 대신 select()를 사용하는 것이었다.

# Before
user = session.query(User).filter(User.id == user_id).first()

# After
from sqlalchemy import select
result = await session.execute(select(User).where(User.id == user_id))
user = result.scalar_one_or_none()

3. 관계 로딩

joinedload를 사용할 때도 문법이 바뀌었다.

from sqlalchemy.orm import selectinload

stmt = select(User).options(selectinload(User.posts)).where(User.id == user_id)
result = await session.execute(stmt)
user = result.scalar_one()

마주친 문제

N+1 쿼리 이슈

기존 코드에서 암묵적으로 처리되던 관계 로딩이 명시적으로 바뀌면서, 미처 selectinload를 추가하지 않은 엔드포인트에서 N+1 문제가 발생했다. 로그를 보고 하나씩 수정했다.

타입 힌트

result.scalar_one()과 result.scalars().all() 등의 반환 타입이 명확하지 않아 mypy 경고가 많이 발생했다. 명시적 타입 캐스팅으로 해결했다.

결과

약 2주간의 마이그레이션 작업 끝에 200여 개의 쿼리를 모두 전환했다. 코드는 더 verbose해졌지만, async/await 패턴으로 통일되어 가독성은 오히려 개선되었다. SQLAlchemy 2.0 정식 릴리즈 시 마이그레이션 부담이 크게 줄어들 것으로 예상한다.