SQLAlchemy N+1 문제 완전 해결 — eager loading 전략

N+1 문제란 무엇인가?

N+1 쿼리 문제는 ORM을 사용하는 개발자라면 반드시 마주치게 되는 성능 함정입니다. 단순히 말하면, 1개의 쿼리로 N개의 레코드를 가져온 뒤, 각 레코드의 연관 데이터를 가져오기 위해 추가로 N번의 쿼리가 실행되는 현상입니다. 예를 들어 게시글 100개를 조회한 뒤 각 게시글의 작성자 정보를 별도로 가져온다면, 총 101번의 쿼리가 실행됩니다. 이는 데이터베이스에 막대한 부하를 주며 응답 속도를 극적으로 저하시킵니다.

SQLAlchemy는 기본적으로 lazy loading 전략을 사용합니다. 연관 객체에 접근할 때 비로소 쿼리를 실행하는 방식이죠. 편리하지만, 반복문 안에서 연관 데이터에 접근하면 N+1 문제가 발생합니다. FastAPI 프로젝트에서 이 문제를 방치하면 수십 배의 성능 저하가 생깁니다.

N+1 문제 재현 — 문제 상황 확인

아래 코드는 전형적인 N+1 문제 상황입니다. Post 목록을 가져온 뒤 각 게시글의 author에 접근하면서 N번의 추가 쿼리가 발생합니다.

# models.py
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, DeclarativeBase

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String(100))
    posts = relationship("Post", back_populates="author")

class Post(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True)
    title = Column(String(200))
    author_id = Column(Integer, ForeignKey("users.id"))
    author = relationship("User", back_populates="posts")
    comments = relationship("Comment", back_populates="post")

class Comment(Base):
    __tablename__ = "comments"
    id = Column(Integer, primary_key=True)
    body = Column(String(500))
    post_id = Column(Integer, ForeignKey("posts.id"))
    post = relationship("Post", back_populates="comments")

# N+1 문제 발생 코드 (나쁜 예시)
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

async def get_posts_bad(db: AsyncSession):
    result = await db.execute(select(Post))
    posts = result.scalars().all()

    # 여기서 N+1 발생!
    # posts가 100개면 author에 접근할 때마다 쿼리 실행
    post_list = []
    for post in posts:
        # lazy load → 개별 SELECT 쿼리 실행
        author = await db.get(User, post.author_id)
        post_list.append({
            "id": post.id,
            "title": post.title,
            "author_name": author.name  # N번 추가 쿼리!
        })
    return post_list

# 실행 결과: posts 100개 → 총 101번 쿼리 실행
# SELECT * FROM posts;                    ← 1번
# SELECT * FROM users WHERE id = 1;      ← N번 (100번)
# SELECT * FROM users WHERE id = 2;
# ... (반복)

해결책 1: selectinload — 가장 많이 쓰는 패턴

selectinload는 SQLAlchemy에서 N+1 문제를 해결하는 가장 일반적인 방법입니다. 부모 쿼리를 먼저 실행한 뒤, 연관 데이터를 IN 절을 사용한 단일 쿼리로 한 번에 가져옵니다. 메모리 효율이 좋고 대부분의 상황에서 안전하게 사용할 수 있습니다.

# selectinload 사용 (올바른 예시)
from sqlalchemy.orm import selectinload
from sqlalchemy import select

async def get_posts_with_selectinload(db: AsyncSession):
    # selectinload로 author와 comments를 한 번에 로드
    stmt = (
        select(Post)
        .options(
            selectinload(Post.author),           # author 일괄 로드
            selectinload(Post.comments)          # comments 일괄 로드
        )
        .order_by(Post.id.desc())
    )
    result = await db.execute(stmt)
    posts = result.scalars().all()

    return [
        {
            "id": post.id,
            "title": post.title,
            "author_name": post.author.name,         # 추가 쿼리 없음!
            "comment_count": len(post.comments)       # 추가 쿼리 없음!
        }
        for post in posts
    ]

# 실행 결과: 총 3번 쿼리 (N과 무관)
# SELECT * FROM posts;
# SELECT * FROM users WHERE id IN (1, 2, 3, ...);   ← 1번
# SELECT * FROM comments WHERE post_id IN (1, 2, 3, ...); ← 1번

해결책 2: joinedload — JOIN으로 한 방에

joinedload는 SQL JOIN을 사용해서 연관 데이터를 부모 쿼리와 함께 한 번에 가져오는 방식입니다. 쿼리 수를 최소화할 수 있지만, 다-대-다(many-to-many) 관계에서는 중복 행이 발생할 수 있어 주의해야 합니다. to-one 관계(ManyToOne, OneToOne)에 특히 적합합니다.

# joinedload 사용 예시
from sqlalchemy.orm import joinedload
from sqlalchemy import select

async def get_posts_with_joinedload(db: AsyncSession):
    stmt = (
        select(Post)
        .options(
            joinedload(Post.author),    # LEFT OUTER JOIN
        )
        .order_by(Post.id.desc())
        .limit(50)
    )
    result = await db.execute(stmt)
    # joinedload 사용 시 unique() 필수!
    posts = result.unique().scalars().all()

    return [
        {
            "id": post.id,
            "title": post.title,
            "author_name": post.author.name,
        }
        for post in posts
    ]

# 실행 SQL (1번 쿼리)
# SELECT posts.*, users.*
# FROM posts
# LEFT OUTER JOIN users ON posts.author_id = users.id
# ORDER BY posts.id DESC
# LIMIT 50;

# ⚠️ 컬렉션(one-to-many)에 joinedload 쓰면 중복 행 발생!
# → 컬렉션은 selectinload 권장, to-one 관계는 joinedload 권장

해결책 3: 중첩 eager loading — 깊은 관계 처리

Post → Comment → User처럼 3단계 이상 중첩된 관계도 eager loading으로 한 번에 처리할 수 있습니다. selectinload를 체이닝하면 됩니다.

# 중첩 eager loading 예시
from sqlalchemy.orm import selectinload, joinedload

async def get_posts_deep(db: AsyncSession):
    stmt = (
        select(Post)
        .options(
            # author: joinedload (to-one)
            joinedload(Post.author),
            # comments: selectinload + 중첩으로 comment의 user도 로드
            selectinload(Post.comments).joinedload(Comment.user),
        )
    )
    result = await db.execute(stmt)
    posts = result.unique().scalars().all()
    return posts

# 실행 결과: 총 2번 쿼리
# 1) SELECT posts.*, users.* FROM posts LEFT JOIN users ...
# 2) SELECT comments.*, users.* FROM comments
#    LEFT JOIN users ON comments.user_id = users.id
#    WHERE comments.post_id IN (...)

FastAPI 실전 패턴 — 의존성 주입과 함께

FastAPI의 의존성 주입 패턴과 결합하면 재사용 가능한 쿼리 함수를 만들 수 있습니다. CRUD 레이어에서 명시적으로 eager loading 전략을 정의하는 것이 권장됩니다.

# crud/post.py
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload, joinedload
from sqlalchemy import select
from typing import Optional

class PostCRUD:
    async def get_list(
        self,
        db: AsyncSession,
        skip: int = 0,
        limit: int = 20,
        load_author: bool = True,
        load_comments: bool = False,
    ) -> list[Post]:
        stmt = select(Post).offset(skip).limit(limit)

        options = []
        if load_author:
            options.append(joinedload(Post.author))
        if load_comments:
            options.append(selectinload(Post.comments))

        if options:
            stmt = stmt.options(*options)

        result = await db.execute(stmt)
        return result.unique().scalars().all()

    async def get_by_id(
        self, db: AsyncSession, post_id: int
    ) -> Optional[Post]:
        stmt = (
            select(Post)
            .where(Post.id == post_id)
            .options(
                joinedload(Post.author),
                selectinload(Post.comments).joinedload(Comment.user),
            )
        )
        result = await db.execute(stmt)
        return result.unique().scalar_one_or_none()

post_crud = PostCRUD()

# router/post.py
from fastapi import APIRouter, Depends
router = APIRouter()

@router.get("/posts")
async def list_posts(
    skip: int = 0,
    limit: int = 20,
    db: AsyncSession = Depends(get_db),
):
    posts = await post_crud.get_list(db, skip=skip, limit=limit)
    return posts

selectinload vs joinedload 선택 기준

어떤 전략을 선택할지는 관계 유형과 데이터 크기에 따라 결정합니다. 다음 기준을 따르면 대부분의 케이스를 커버할 수 있습니다.

• joinedload → ManyToOne, OneToOne 관계 (단일 연관 객체). SQL JOIN으로 처리, 쿼리 수 최소화.
• selectinload → OneToMany, ManyToMany 관계 (컬렉션). 중복 행 없이 안전하게 처리.
• 중첩 selectinload → 깊은 관계 그래프가 필요할 때. 체이닝으로 명확하게 표현.
• raiseload → 예상치 못한 lazy load를 방지하고 싶을 때 (개발/테스트 환경에서 유용).

raiseload로 N+1 방어망 구축

raiseload는 실수로 lazy loading이 발생하면 즉시 예외를 던집니다. 개발 환경에서 N+1을 조기에 잡아내는 데 매우 유용합니다.

# raiseload로 lazy loading 방어
from sqlalchemy.orm import raiseload

async def get_posts_safe(db: AsyncSession):
    stmt = (
        select(Post)
        .options(
            joinedload(Post.author),     # 명시적 eager load
            raiseload("*"),              # 나머지 모든 관계 접근 시 예외 발생
        )
    )
    result = await db.execute(stmt)
    posts = result.unique().scalars().all()

    # post.comments에 접근하면 즉시 MissingGreenlet 예외!
    # → 놓친 N+1을 개발 단계에서 발견할 수 있음
    return posts

# 모델 레벨에서 기본 전략을 eager로 설정하는 방법
class Post(Base):
    __tablename__ = "posts"
    # ...
    author = relationship(
        "User",
        back_populates="posts",
        lazy="joined"    # 항상 joinedload 사용
    )
    comments = relationship(
        "Comment",
        back_populates="post",
        lazy="selectin"  # 항상 selectinload 사용
    )

EXPLAIN ANALYZE로 쿼리 검증

실제 쿼리가 몇 번 실행되는지 확인하려면 SQLAlchemy 이벤트 리스너나 로깅을 활용하세요. 개발 환경에서는 반드시 쿼리 카운트를 모니터링해야 합니다.

# 쿼리 로깅 설정 (개발 환경)
import logging

# SQLAlchemy 쿼리 로그 활성화
logging.basicConfig()
logging.getLogger("sqlalchemy.engine").setLevel(logging.INFO)

# 또는 pytest에서 쿼리 수 검증
from sqlalchemy import event

class QueryCounter:
    def __init__(self, engine):
        self.count = 0
        self.engine = engine
        event.listen(engine.sync_engine, "before_cursor_execute", self._count)

    def _count(self, *args, **kwargs):
        self.count += 1

# pytest 테스트 예시
async def test_no_n_plus_one(db, engine):
    counter = QueryCounter(engine)
    posts = await post_crud.get_list(db, limit=100)
    _ = [p.author.name for p in posts]  # author 접근

    # 100개 게시글이 있어도 쿼리는 최대 2번이어야 함
    assert counter.count <= 2, f"N+1 감지! {counter.count}번 쿼리 실행됨"

정리: N+1 방지 체크리스트

• 반복문 안에서 연관 객체에 접근하는 코드가 있다면 즉시 selectinload/joinedload 적용
• to-one 관계는 joinedload, 컬렉션(one-to-many)은 selectinload
• 개발 환경에서 raiseload와 쿼리 카운터로 N+1 자동 감지
• 프로덕션에서는 slow query log와 APM 도구로 모니터링
• 필요한 관계만 로드 (불필요한 eager load도 성능을 낮춤)

코드벤터는 FastAPI + SQLAlchemy 기반의 서비스를 다수 구축하면서 쌓인 쿼리 최적화 노하우를 글로벌 협력 네트워크와 함께 공유합니다. N+1 문제 하나만 제대로 잡아도 서비스 응답속도가 극적으로 개선되는 경험을 해보시길 바랍니다. 더 궁금한 점은 코드벤터 블로그나 댓글로 문의해 주세요.