SQLAlchemy 2.0 #6 쿼리 심화: 조인, 집계, 서브쿼리, 대량 처리

프로그래밍 언어 파이썬 SQLAlchemy 데이터베이스
5 분 소요

모델과 세션, 관계까지 갖췄으니 이제 쿼리의 표현력을 넓힐 차례입니다. 2.0의 좋은 점은 여기서 새 문법이 나오지 않는다는 것입니다. 전부 select() 위에 조립되고, Core에서든 ORM에서든 같은 모양입니다. 이번 편은 실무에서 반복해서 쓰게 되는 패턴 모음입니다. 예제는 5편의 User, Address 모델을 계속 씁니다.

scalars vs execute: 결과를 받는 두 형태 #

먼저 결과 받는 법부터 정리해야 헷갈리지 않습니다.

main.py 
from sqlalchemy import select

# 엔티티 하나를 조회할 때: scalars()
users = session.scalars(select(User).where(User.name.like("김%"))).all()
# → [User, User, ...] 객체 리스트

# 여러 값을 조회할 때: execute()
rows = session.execute(select(User.name, User.email)).all()
# → [('김파이썬', 'kim@example.com'), ...] Row 리스트
for name, email in rows:
    print(name, email)

규칙은 하나입니다. SELECT 대상이 엔티티 하나면 scalars(), 컬럼 여러 개나 엔티티 + 집계의 혼합이면 execute()를 씁니다. execute()로 엔티티 하나를 조회하면 (User,)처럼 원소 하나짜리 Row에 싸여 나오는데, 초심자가 가장 자주 걸려 넘어지는 지점입니다.

단건 조회에는 전용 메서드가 있습니다.

main.py 
user = session.get(User, 1)                  # 기본 키 조회. 없으면 None
user = session.scalars(stmt).first()         # 첫 행 또는 None
user = session.scalars(stmt).one()           # 정확히 1행이 아니면 예외

one()은 “반드시 하나여야 하는” 조회(고유 조건 조회)에 씁니다. 0건이든 2건이든 예외가 나므로 데이터 이상을 조기에 발견합니다.

조건 조합: and, or, in #

main.py 
from sqlalchemy import or_

stmt = select(User).where(
    User.name.like("김%"),                    # where 나열 = AND
    User.id.in_([1, 2, 3]),
)

stmt = select(User).where(
    or_(User.name == "김파이썬", User.name == "이코어")
)

where()에 조건을 나열하면 AND로 묶입니다. OR만 or_()로 감쌉니다. in_()에는 리스트뿐 아니라 서브쿼리도 들어갑니다(아래에서 다룹니다).

조인: relationship이 ON 절을 알고 있습니다 #

main.py 
# ORM 조인: ON 절은 relationship 선언에서 추론
stmt = (
    select(User.name, Address.email_address)
    .join(User.addresses)
    .where(Address.email_address.like("%@work.com"))
)

# 관계 선언이 없거나 모호할 때: ON 절 명시
stmt = select(User.name, Address.email_address).join(
    Address, User.id == Address.user_id
)

# LEFT OUTER JOIN: 주소가 없는 사용자도 포함
stmt = select(User.name, Address.email_address).join(User.addresses, isouter=True)

join(User.addresses)처럼 관계 속성을 넘기면 ON 절을 쓸 필요가 없습니다. 5편의 eager loading(selectinload)과 헷갈리기 쉬운데, 역할이 다릅니다. join()은 SQL의 WHERE, SELECT에 다른 테이블을 쓰기 위한 것이고, selectinload()는 관계 속성을 미리 채워 두기 위한 것입니다. “주소가 work.com인 사용자를 찾고 싶다"는 join, “사용자 목록과 각자의 주소를 다 보여 주고 싶다"는 selectinload입니다.

집계: group_by와 label #

main.py 
from sqlalchemy import func

stmt = (
    select(User.name, func.count(Address.id).label("address_count"))
    .join(User.addresses, isouter=True)
    .group_by(User.id)
    .having(func.count(Address.id) >= 2)
    .order_by(func.count(Address.id).desc())
)
for row in session.execute(stmt):
    print(row.name, row.address_count)  # label 덕에 이름으로 접근
  • func.아무이름()은 해당 이름의 SQL 함수를 그대로 호출합니다. func.count, func.sum, func.max는 물론 DB 고유 함수도 됩니다.
  • label()을 붙이면 결과 Row에서 그 이름으로 읽을 수 있습니다. 집계 컬럼에는 습관적으로 붙이는 것이 좋습니다.
  • 집계 결과에 조건을 걸 때는 where가 아니라 having입니다. SQL 규칙 그대로입니다.

서브쿼리와 EXISTS #

“주소가 하나라도 있는 사용자"는 두 가지로 쓸 수 있습니다.

main.py 
# IN + 서브쿼리
subq = select(Address.user_id)
stmt = select(User).where(User.id.in_(subq))

# EXISTS: 상관 서브쿼리
from sqlalchemy import exists

stmt = select(User).where(
    exists().where(Address.user_id == User.id)
)

집계 결과를 조인에 쓰려면 subquery()로 이름 붙은 파생 테이블을 만듭니다.

main.py 
addr_count = (
    select(Address.user_id, func.count(Address.id).label("cnt"))
    .group_by(Address.user_id)
    .subquery()
)
stmt = (
    select(User.name, addr_count.c.cnt)
    .join(addr_count, User.id == addr_count.c.user_id)
)

서브쿼리의 컬럼은 subq.c.컬럼명으로 접근합니다. Core의 Table.c와 같은 인터페이스입니다.

페이지네이션: LIMIT·OFFSET과 그 한계 #

main.py 
page, per_page = 3, 20
stmt = (
    select(User)
    .order_by(User.id)                 # 순서 고정 없이는 페이지가 뒤섞입니다
    .limit(per_page)
    .offset((page - 1) * per_page)
)

두 가지를 기억해야 합니다. 첫째, order_by 없는 페이지네이션은 정의되지 않은 동작입니다. DB는 순서를 보장하지 않으므로 페이지마다 행이 겹치거나 빠질 수 있습니다. 둘째, OFFSET은 건너뛰는 행도 일단 읽습니다. 깊은 페이지(OFFSET 100000)는 그만큼 느려지므로, 무한 스크롤류에는 “마지막으로 본 id보다 큰 것"을 조건으로 거는 키셋(keyset) 방식이 낫습니다.

main.py 
# 키셋 페이지네이션: 깊이와 무관하게 일정한 속도
stmt = select(User).where(User.id > last_seen_id).order_by(User.id).limit(per_page)

대량 처리: 단위 작업을 우회하기 #

수만 건을 session.add()로 넣으면 변경 추적 비용이 그대로 쌓입니다. 대량 작업은 ORM의 편의를 포기하고 표현식을 직접 실행하는 것이 맞습니다.

main.py 
from sqlalchemy import insert, update

# 대량 INSERT: 딕셔너리 리스트로 executemany
session.execute(
    insert(User),
    [{"name": f"user{i}", "email": f"user{i}@example.com"} for i in range(10_000)],
)

# 대량 UPDATE: 조건에 걸리는 모든 행을 한 문장으로
session.execute(
    update(User).where(User.name.like("test%")).values(name="정리됨")
)
session.commit()

이 방식은 객체를 만들지 않고 세션의 identity map도 거치지 않습니다. 그래서 빠르지만, 이미 세션에 로드된 객체에는 변경이 자동 반영되지 않는다는 대가가 있습니다. 대량 UPDATE 후 같은 세션에서 그 행들을 다시 쓸 일이 있다면 커밋 후 재조회가 안전합니다.

정리 #

  • 엔티티 하나는 scalars(), 컬럼 혼합은 execute()로 받습니다. 단건은 get(), first(), one()을 용도별로 씁니다.
  • 조인은 관계 속성을 넘기면 ON 절이 추론됩니다. 필터용은 join(), 관계 속성 로드용은 selectinload()로 역할이 다릅니다.
  • 집계는 func + label + group_by, 집계 조건은 having입니다. 서브쿼리는 in_(), exists(), subquery()로 조립합니다.
  • 페이지네이션은 order_by가 필수이고, 깊은 페이지는 OFFSET 대신 키셋 방식을 씁니다.
  • 대량 INSERT·UPDATE는 세션의 변경 추적을 우회해 표현식으로 직접 실행합니다. 로드된 객체와의 불일치만 주의합니다.
  • 마지막 편에서는 스키마 변경을 관리하는 Alembic, 비동기 지원, 실전 프로젝트 구성을 다룹니다.
X