FastAPI에서 SQLAlchemy 2.0 스타일 쿼리 적용기

배경

회사 API 서버가 FastAPI + SQLAlchemy 1.4를 사용 중이다. SQLAlchemy 2.0이 곧 나온다는 소식에 마이그레이션을 대비해 1.4의 2.0 스타일을 미리 적용하기로 했다.

기존 코드는 Query 객체를 직접 사용하는 레거시 방식이었다.

# 기존 방식
user = db.query(User).filter(User.id == user_id).first()

변경 사항

2.0 스타일은 select() 함수를 사용한다.

from sqlalchemy import select

# 2.0 스타일
stmt = select(User).where(User.id == user_id)
result = db.execute(stmt)
user = result.scalar_one_or_none()

처음엔 scalar(), scalars(), scalar_one() 등 결과 처리 메서드가 헷갈렸다.

  • scalar(): 첫 번째 row의 첫 컬럼
  • scalars(): 모든 row의 첫 컬럼 (iterable)
  • scalar_one(): 정확히 1개, 없거나 여러 개면 예외
  • scalar_one_or_none(): 0개 또는 1개

조인 쿼리 변경

조인 쿼리도 문법이 달라진다.

# 기존
results = db.query(User, Post).join(Post).all()

# 2.0
stmt = select(User, Post).join(Post)
results = db.execute(stmt).all()

설정 추가

future=True 플래그를 엔진과 세션에 추가해야 2.0 동작이 활성화된다.

engine = create_engine(DATABASE_URL, future=True)
SessionLocal = sessionmaker(bind=engine, future=True)

마이그레이션 전략

전체 코드를 한 번에 바꾸긴 어려워서 신규 엔드포인트부터 적용하고, 레거시는 점진적으로 수정하는 방향으로 진행 중이다. 1.4는 두 방식을 모두 지원하니 혼용 가능하다.

타입 힌팅도 개선되어 있어서 mypy 체크가 더 정확해진 점도 장점이다.

FastAPI에서 SQLAlchemy 2.0 스타일 쿼리 적용기