Django上級 #3 クエリ最適化 — N+1、select_related、EXPLAIN、インデックス

フレームワーク Django ORM パフォーマンス
読了 9分

中級 #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 1 回で解決。

✅ 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 まで 1 クエリで付いてきます。ただし JOIN が増えると結果の行幅が大きくなるので 3-4 段以上は慎重に。

select_related が効くケース:

  • ForeignKey (自分が持っている側)
  • OneToOneField (両側)

効かないケース:

  • 逆方向 ForeignKey (author.posts)
  • ManyToMany

この 2 つは 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 回

Django が 最初のクエリの ID を集めて IN 句で 2 つ目のクエリ を投げ、Python でマッピングします。逆方向のコレクションが N+1 なしで埋まります。

select_relatedprefetch_related
方式SQL JOIN別クエリ + Python マッチ
クエリ数+0+1 (関係ごと)
似合う場面順方向 FK、1:1逆方向 FK、M2M
結果の幅行が広くなるそのまま

チェーンも一緒に。

2 つを混ぜる 
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()

Django は recent_ids を自動でサブクエリに落として 1 回の SQL にまとめます。

計測 #

計測なくして最適化なし。 Django には道具がよく揃っています。

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)

実際に 1 回実行して行数、時間、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 — メモリ節約 #

🚫 100 万行をメモリに 
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()

2 つのリクエストが同時に来ると、両方とも 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()

行単位のロック がトランザクションが終わるまで保持されます。2 番目のリクエストは 1 番目が 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)

デッドロック回避 #

デッドロックは 2 つのトランザクションがお互いのロックを待っている 状態。回避ルール:

  • 常に同じ順序で ロック取得 (例: 常に小さい 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()。

ただし すでに結果がキャッシュされた QuerySet なら len は追加クエリなしで動作 — 場面による。

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

🚫 全部数える 
if Post.objects.filter(slug=s).count() > 0: ...
✅ 1 行だけ確認 
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 まで — クエリを減らすもう 1 つの軸。

X