장고 고급 #3 쿼리 최적화 — N+1, select_related, EXPLAIN, 인덱스

프레임워크 장고 ORM 성능
7 분 소요

중급 #2 ORM 중급에서 다진 쿼리 위에 본격 최적화를 얹습니다. 운영에서 99% 의 성능 문제는 ORM 사용 패턴에서 옵니다. CPU도, 메모리도 아니고 쿼리 수와 형태.

이번 글의 도구상자:

  • N+1 진단과 해결 (select_related, prefetch_related, Prefetch)
  • 필드 선택 (only, defer, values, values_list)
  • 측정 (debug-toolbar, connection.queries, .explain())
  • 인덱스 (Meta.indexes, db_index, 멀티컬럼)
  • 대량 작업 (bulk_create, bulk_update, update_or_create, iterator)
  • 동시성 (select_for_update)
  • 출구 (raw SQL)

N+1 — 가장 흔한 함정 #

재현 #

models.py 
class Author(models.Model):
    name = models.CharField(max_length=100)

class Post(models.Model):
    title = models.CharField(max_length=200)
    author = models.ForeignKey(Author, on_delete=models.CASCADE, related_name="posts")
🚫 N+1 발생 
posts = Post.objects.all()[:50]
for post in posts:
    print(post.title, post.author.name)   # ← 여기서 매번 추가 쿼리

쿼리 흐름:

SELECT * FROM post LIMIT 50;                      -- 1번
SELECT * FROM author WHERE id = ?;                -- 50번
-- 총 51번

50개 게시물 페이지가 51번 쿼리. 게시물이 1000개면 1001번. 이게 N+1.

select_related — 정방향 FK / OneToOne #

JOIN 한 번으로 해결.

✅ select_related 
posts = Post.objects.select_related("author").all()[:50]
for post in posts:
    print(post.title, post.author.name)

쿼리:

SELECT post.*, author.*
FROM post LEFT OUTER JOIN author ON post.author_id = author.id
LIMIT 50;
-- 1번

체인도 됩니다.

여러 단계 
Comment.objects.select_related("post__author__profile").all()

comment.post.author.profile까지 한 쿼리에 따라옵니다. 다만 JOIN이 늘어나면 결과 행 폭이 커지니 3-4 단계 이상은 신중히.

select_related가 통하는 경우:

  • ForeignKey (자기가 가진 쪽)
  • OneToOneField (양쪽)

통하지 않는 경우:

  • 역방향 ForeignKey (author.posts)
  • ManyToMany

이 둘은 prefetch_related의 쓰임입니다.

prefetch_related — 역방향 FK / M2M #

별도 쿼리 + Python 측 조인.

✅ prefetch_related 
authors = Author.objects.prefetch_related("posts").all()[:20]
for author in authors:
    print(author.name, [p.title for p in author.posts.all()])

쿼리 흐름:

SELECT * FROM author LIMIT 20;                                    -- 1번
SELECT * FROM post WHERE author_id IN (1, 2, ..., 20);            -- 1번
-- 총 2번

장고가 첫 쿼리의 ID 들을 모아 IN 절로 두 번째 쿼리를 던지고, Python에서 매핑합니다. 역방향 컬렉션이 N+1 없이 채워집니다.

select_relatedprefetch_related
방식SQL JOIN별도 쿼리 + 파이썬 매칭
쿼리 수+0+1 (per 관계)
어울리는 경우정방향 FK, 1:1역방향 FK, M2M
결과 폭행이 넓어짐그대로

체인도 같이.

둘을 섞기 
Comment.objects.select_related("author").prefetch_related("post__tags")

Prefetch 객체 — 커스터마이즈 #

prefetch 되는 쿼리에 추가 조건이나 ordering을 주고 싶을 때.

Prefetch 
from django.db.models import Prefetch

authors = Author.objects.prefetch_related(
    Prefetch(
        "posts",
        queryset=Post.objects.filter(published=True).order_by("-created_at"),
        to_attr="recent_posts",   # author.recent_posts로 접근
    )
).all()

for author in authors:
    for post in author.recent_posts:
        ...

to_attr로 원래 manager와 분리된 속성에 채울 수 있습니다. author.posts.all()은 그대로 모든 글, author.recent_posts는 필터된 글.

필드 선택 — only, defer, values, values_list #

모델에 컬럼이 30개 있는데 5개만 쓰면 70% 의 데이터가 낭비입니다.

only — 명시한 필드만 즉시 로드 #

only 
posts = Post.objects.only("id", "title").all()
for p in posts:
    print(p.title)        # OK, 안 가져온 필드 접근 시 추가 쿼리
    print(p.body)         # ← 추가 쿼리 발생 (deferred 필드)

body같이 큰 텍스트 컬럼이 있고 목록 페이지에서 안 쓰면 큰 효과.

defer — 명시한 필드만 미루기 #

defer — only의 반대 
posts = Post.objects.defer("body", "raw_html").all()

values, values_list — ORM 객체 안 만들기 #

ORM 객체 인스턴스화 비용까지 빼고 dict / tuple로.

values / values_list 
# dict의 리스트
Post.objects.values("id", "title")
# [{"id": 1, "title": "..."}, ...]

# tuple의 리스트
Post.objects.values_list("id", "title")
# [(1, "..."), ...]

# 단일 컬럼 평탄화
ids = list(Post.objects.values_list("id", flat=True))
# [1, 2, 3, ...]

ID만 모은다거나, JSON 응답으로 그대로 직렬화한다거나, CSV 익스포트 같은 경우에 결정적.

실용 — IN 절을 위한 ID 추출 
recent_ids = Post.objects.filter(
    created_at__gte=cutoff
).values_list("id", flat=True)

Comment.objects.filter(post_id__in=recent_ids).delete()

장고는 recent_ids를 자동으로 서브쿼리로 내려서 한 번의 SQL로 합칩니다.

측정 #

측정 없이 최적화 없다. 장고에는 도구가 잘 갖춰져 있습니다.

django-debug-toolbar #

개발 환경의 표준.

설치 
pip install django-debug-toolbar
settings.py 
INSTALLED_APPS = [..., "debug_toolbar"]
MIDDLEWARE = [..., "debug_toolbar.middleware.DebugToolbarMiddleware"]
INTERNAL_IPS = ["127.0.0.1"]
urls.py 
from django.conf import settings
from django.urls import include, path

urlpatterns = [...]

if settings.DEBUG:
    urlpatterns += [path("__debug__/", include("debug_toolbar.urls"))]

페이지 우측 패널에 쿼리 수, 각 쿼리, EXPLAIN, 시간이 다 보입니다. 쿼리 수가 페이지 항목 수에 비례하면 N+1의심.

connection.queries #

코드 안에서 직접 보고 싶을 때.

DEBUG=True 일 때만 
from django.db import connection, reset_queries

reset_queries()
posts = list(Post.objects.select_related("author")[:50])
for p in posts:
    _ = p.author.name
print(len(connection.queries), "queries")
for q in connection.queries:
    print(q["time"], q["sql"][:100])

DEBUG=True 또는 CONN_MAX_AGE가 활성일 때만 채워집니다. 테스트 코드에서는 assertNumQueries가 더 표준.

테스트로 보장 
from django.test import TestCase

class PostListQueryCountTest(TestCase):
    def test_no_n_plus_one(self):
        for i in range(10):
            author = Author.objects.create(name=f"A{i}")
            Post.objects.create(title=f"P{i}", author=author)

        with self.assertNumQueries(2):   # author + posts
            list(self.client.get("/posts/").context["posts"])

쿼리 수에 대한 회귀 테스트가 N+1의 재발 방지에 가장 효과적.

.explain() #

DB가 실제로 쓸 실행 계획을 본다.

explain 
qs = Post.objects.filter(published=True, author__name__startswith="K")
print(qs.explain())

PostgreSQL 출력 일부:

Hash Join  (cost=15.50..120.30 rows=200 width=64)
  Hash Cond: (post.author_id = author.id)
  ->  Seq Scan on post  (cost=0.00..80.00 rows=2000 width=40)
        Filter: published
  ->  Hash  (cost=12.00..12.00 rows=200 width=24)
        ->  Seq Scan on author  (cost=0.00..12.00 rows=200 width=24)
              Filter: ((name)::text ~~ 'K%'::text)

Seq Scan이 큰 테이블에 보이면 인덱스 누락 신호. Index Scan이 들어오는지 인덱스 추가 후 다시 explain.

PostgreSQL만의 옵션:

ANALYZE 까지 
qs.explain(analyze=True, verbose=True, buffers=True)

실제로 한 번 실행해서 행 수, 시간, IO까지 보여줍니다. 운영 비슷한 데이터에서 돌려야 의미가 있습니다.

인덱스 #

db_index=True — 단일 컬럼 #

컬럼 단위 
class Post(models.Model):
    slug = models.SlugField(max_length=200, unique=True)   # unique 면 자동 인덱스
    status = models.CharField(max_length=20, db_index=True)
    created_at = models.DateTimeField(db_index=True)

unique=True, primary_key=True, ForeignKey는 자동으로 인덱스가 붙습니다.

Meta.indexes — 멀티컬럼, 함수, 부분 인덱스 #

Meta.indexes 
from django.db.models import Index, F, Q

class Post(models.Model):
    status = models.CharField(max_length=20)
    created_at = models.DateTimeField()
    author = models.ForeignKey("Author", on_delete=models.CASCADE)
    title = models.CharField(max_length=200)

    class Meta:
        indexes = [
            # 멀티컬럼 (status, created_at) — 순서 중요
            Index(fields=["status", "-created_at"], name="post_status_recent_idx"),

            # 함수 인덱스 (PostgreSQL)
            Index(F("title").lower(), name="post_title_lower_idx"),

            # 부분 인덱스 (조건부)
            Index(
                fields=["created_at"],
                condition=Q(status="published"),
                name="post_published_recent_idx",
            ),
        ]

룰:

  • 멀티컬럼 인덱스의 컬럼 순서는 WHERE 절 순서와 맞춰야 효과
  • 인덱스 추가는 쓰기 비용이 늘어남 — 모든 컬럼에 붙이지 말 것
  • 부분 인덱스는 자주 조회하는 부분만 — 크기 작고 빠름

인덱스가 안 통하는 경우 #

함수가 컬럼을 감싸면 일반 인덱스는 못 씁니다.

🚫 인덱스 못 씀 
WHERE LOWER(title) = 'hello'

해결: 함수 인덱스 (Index(F("title").lower(), ...))를 만들거나, 데이터 측에서 미리 정규화 (저장 시 소문자로).

LIKE 'hello%'는 인덱스 OK, LIKE '%hello'는 인덱스 못 씀 (앞이 와일드). 풀텍스트가 필요하면 PostgreSQL의 SearchVector / GIN 인덱스로.

대량 작업 #

bulk_create — 한 번에 INSERT #

🚫 1000 번 INSERT 
for data in items:
    Post.objects.create(**data)
✅ 1번 INSERT 
posts = [Post(**data) for data in items]
Post.objects.bulk_create(posts, batch_size=500)

수백 배 빠릅니다. 다만 한계:

  • 자동 PK가 채워지는 것은 PostgreSQL만 보장 (다른 DB는 None 인 경우 있음)
  • pre_save / post_save 시그널이 호출되지 않음 (#5)
  • M2M은 별도로 처리

bulk_update — 한 번에 UPDATE #

bulk_update 
posts = list(Post.objects.filter(status="draft"))
for p in posts:
    p.status = "published"
Post.objects.bulk_update(posts, ["status"], batch_size=500)

업데이트할 필드를 명시. save()가 일으키는 시그널은 안 일어납니다.

update_or_create — 멱등 upsert #

upsert 패턴 
DailyStat.objects.update_or_create(
    date=target,
    defaults={"order_count": n, "total_amount": total},
)

#2의 통계 커맨드에서 본 그것. 재실행해도 안전.

PostgreSQL의 진짜 INSERT … ON CONFLICT는 5.x부터 bulk_create(..., update_conflicts=True)로 지원:

bulk_create의 upsert 
Post.objects.bulk_create(
    posts,
    update_conflicts=True,
    update_fields=["title", "body"],
    unique_fields=["slug"],
)

iterator — 메모리 절약 #

🚫 백만 행을 메모리에 
for post in Post.objects.all():
    process(post)
✅ 청크 단위 
for post in Post.objects.iterator(chunk_size=2000):
    process(post)

iterator는 결과를 캐시하지 않고 DB 커서로 스트리밍합니다. 큰 테이블 순회의 표준.

동시성 — select_for_update #

같은 행을 여러 워커가 동시에 갱신하는 경우입니다. 잔액 차감, 재고 감소 등.

🚫 race condition 
def withdraw(account_id, amount):
    account = Account.objects.get(pk=account_id)
    if account.balance < amount:
        raise ValueError("부족")
    account.balance -= amount
    account.save()

두 요청이 동시에 들어오면 둘 다 get 시점에 잔액을 보고 통과한 뒤 둘 다 차감 — 잔액이 음수가 됩니다.

✅ select_for_update + atomic 
from django.db import transaction

def withdraw(account_id, amount):
    with transaction.atomic():
        account = Account.objects.select_for_update().get(pk=account_id)
        if account.balance < amount:
            raise ValueError("부족")
        account.balance -= amount
        account.save()

행 단위 락이 트랜잭션이 끝날 때까지 잡힙니다. 두 번째 요청은 첫 번째가 commit/rollback 할 때까지 대기.

nowait, skip_locked #

대기 안 하고 다른 행으로 
Job.objects.select_for_update(skip_locked=True).filter(status="pending").first()

Job 큐 패턴에 자주 등장 — 다른 워커가 잡은 행은 건너뛰고 다음 일감을 잡습니다.

대기 안 하고 즉시 실패 
account = Account.objects.select_for_update(nowait=True).get(pk=id)

데드락 회피 #

데드락은 두 트랜잭션이 서로의 락을 기다리는 상태. 회피 룰:

  • 항상 같은 순서로 락 획득 (예: 항상 작은 ID 먼저)
  • 트랜잭션을 짧게 — 외부 API 호출을 트랜잭션 안에 넣지 말 것
  • 필요한 행만 락 — select_for_update().filter(...)를 좁게

Raw SQL — 출구 #

ORM으로 표현하기 너무 복잡하거나 DB 고유 기능이 필요할 때.

Manager.raw 
posts = Post.objects.raw(
    "SELECT * FROM post WHERE to_tsvector('simple', title) @@ to_tsquery('simple', %s)",
    ["python | django"],
)
for p in posts:
    print(p.title)

raw는 ORM 객체로 결과를 매핑해줘서, 모델의 메소드/프로퍼티가 살아 있습니다.

더 낮은 레이어:

connection.cursor 
from django.db import connection

with connection.cursor() as cur:
    cur.execute(
        "UPDATE post SET status = %s WHERE created_at < %s RETURNING id",
        ["archived", cutoff],
    )
    archived_ids = [row[0] for row in cur.fetchall()]

ORM와 분리된 경로입니다. SQL 인젝션 방지를 위해 항상 파라미터 바인딩 (%s)을 쓸 것 — 절대 Python f-string으로 SQL을 만들지 마세요.

자주 만나는 함정 #

1) len(qs)와 qs.count() #

🚫 모두 가져와서 세기 
n = len(Post.objects.all())
✅ COUNT(*) 
n = Post.objects.count()

len은 결과를 다 메모리에 올려서 셉니다. 카운트만 필요하면 count().

다만 이미 결과가 캐시된 큐리셋 이라면 len이 추가 쿼리 없이 동작 — 상황에 따라.

2) exists() vs count() > 0 #

🚫 다 세기 
if Post.objects.filter(slug=s).count() > 0: ...
✅ 한 행만 확인 
if Post.objects.filter(slug=s).exists(): ...

exists는 LIMIT 1로 끝납니다.

3) .first()로 NULL 체크 #

✅ 깔끔 
post = Post.objects.filter(slug=s).first()
if post is None:
    return Http404

get + try/except DoesNotExist보다 짧고 의도 명확.

4) M2M 쿼리 #

prefetch_related 없이 for tag in post.tags.all()를 N 개의 post에 돌리면 N+1.

5) count()는 캐시되지 않음 #

같은 트랜잭션에서도 count() 호출은 매번 SQL 입니다. 같은 카운트를 여러 번 쓴다면 변수에 담으세요.

정리 #

이번 글에서 잡은 것:

  • N+1 — 정방향은 select_related (JOIN), 역방향/M2M은 prefetch_related (별도 쿼리)
  • Prefetch(queryset=..., to_attr=...)로 prefetched 결과 커스터마이즈
  • only, defer로 컬럼 선택, values/values_list(flat=True)로 ORM 인스턴스화 회피
  • 측정: debug-toolbar, assertNumQueries, connection.queries, .explain()
  • 인덱스: db_index, Meta.indexes (멀티컬럼/함수/부분), 멀티컬럼은 순서 중요
  • 대량: bulk_create, bulk_update, update_or_create, iterator(chunk_size=...)
  • 동시성: select_for_update (+ atomic), skip_locked (Job 큐), 데드락 회피
  • Raw 출구: Manager.raw, connection.cursor — 파라미터 바인딩 필수
  • 함정: len vs count, count > 0 vs exists, M2M, count 캐시 없음

다음 글(#4 캐싱)에서는 이 위에 캐시를 얹습니다. per-view, template fragment, low-level, Redis 백엔드, 무효화 전략, stampede까지 — 쿼리를 줄이는 또 다른 한 축을 다룹니다.

X