[FastAPI] N+1 Problem

1. N+1 Problem

ORM으로 관계형 데이터를 가져올 때, 한번의 쿼리(N) 로 가져온 각 항목에 대해 추가적인 쿼리 (+1)가 반복 실행되는 문제

예시

• 블로그 글(Post) 10개를 가져오면 (SELECT * FROM post)
• 각 글의 작성자(User)를 가져오기 위해 추가로 10개의 쿼리 (SELECT * FROM user WHERE id = ?) 실행됨

 

 

2. FastAPI + SQLAlchemy 예제

모델 정의

# models.py
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship
from app.database import Base

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String)

class Post(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String)
    user_id = Column(Integer, ForeignKey("users.id"))
    author = relationship("User", backref="posts")

 

# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base, Session

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"  # 예시용 SQLite

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}  # SQLite 전용 설정
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

def get_db():
    db: Session = SessionLocal()
    try:
        yield db
    finally:
        db.close()

 

 

문제 코드

# routers/post_router.py
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session, joinedload
from app.database import get_db
from app.models import Post
import time

router = APIRouter()

@router.get("/posts_nplus1")
def get_posts_nplus1(db: Session = Depends(get_db)):
    start_time = time.perf_counter()

    # author는 lazy-loading → post 개수만큼 쿼리 발생 (N+1)
    posts = db.query(Post).all()

    result = [
        {
            "id": post.id,
            "title": post.title,
            "author": post.author.name  # ❌ 여기서 매번 쿼리 발생
        }
        for post in posts
    ]

    end_time = time.perf_counter()
    print(f"❌ N+1 처리 시간: {end_time - start_time:.4f}초")
    return result

 

 

해결 코드 ( N+1 방지 - joinedload 사용)

@router.get("/posts_optimized")
def get_posts_optimized(db: Session = Depends(get_db)):
    start_time = time.perf_counter()

    # author를 미리 join → 1번 쿼리로 모두 가져옴
    posts = db.query(Post).options(joinedload(Post.author)).all()

    result = [
        {
            "id": post.id,
            "title": post.title,
            "author": post.author.name  # ✅ 이미 로드됨
        }
        for post in posts
    ]

    end_time = time.perf_counter()
    print(f"✅ Optimized 처리 시간: {end_time - start_time:.4f}초")
    return

 

 

결과

User - 2000명, Post - 20000개 SQLite 기준

1. N+1 문제 코드 터미너

N+1 문제가 발생하여 수많은 쿼리 로그가 쌓인다.

 

 

2. join 쳐서 가져오는 경우

한번의 쿼리로 join을 쳐서 데이터를 가져온다.

 

 

호출 경로 함수	평균 처리 시간
posts_nplus1	0.3136초
posts_optimized	0.1557초