Django中級 #2 ORM 中級 — annotate、aggregate、F/Q、prefetch_related

フレームワーク Django ORM
読了 7分

基礎 #3 で ORM の最初の挨拶 — Post.objects.filter、get、create のようなツールを固めました。今回はその上に 本格的なツール を積み上げます。実務でほぼ毎日使うパターンで、この 1 編で ORM で書くコードの半分が変わります。

  • annotate / aggregate — 集計
  • F 式 — DB 側の演算
  • Q オブジェクト — 複合条件 (OR/AND/NOT)
  • select_related / prefetch_related — N+1 問題の解決
  • only / defer / values — 取得するフィールドの選択
  • raw SQL の出口 — 最後のカード

前回の #1 CBV で見た get_queryset の中に入れるツールでもあります。

annotate と aggregate — 集計の 2 つの顔 #

両方とも集計関数を使いますが、結果の形 が異なります。

annotateaggregate
結果QuerySet (各行に新フィールド)dict (全体に対する集計値)
用途行単位の計算全体単位の計算
例“各記事のコメント数”“全体の記事数、平均評価”

aggregate — 全体に対する集計 #

aggregate 
from django.db.models import Count, Avg, Sum, Max, Min

Post.objects.aggregate(Count("id"))
# {'id__count': 1234}

Post.objects.aggregate(
    total=Count("id"),
    avg_views=Avg("views"),
    max_views=Max("views"),
)
# {'total': 1234, 'avg_views': 152.3, 'max_views': 9876}

キー名は自動生成されるか (<field>__<func> 形式)、キーワード引数で直接指定できます。

annotate — 行単位の計算 #

annotate — 各記事のコメント数 
from django.db.models import Count

posts = Post.objects.annotate(comment_count=Count("comments"))
for p in posts:
    print(p.title, p.comment_count)

comments は Comment モデルの ForeignKey(Post, related_name="comments") の逆方向マネージャ名です。annotate が SQL GROUP BY + 集計関数を自動で作り、各記事ごとにコメント数を計算します。

生成される SQL (簡略化) 
SELECT post.*, COUNT(comment.id) AS comment_count
FROM post
LEFT OUTER JOIN comment ON comment.post_id = post.id
GROUP BY post.id;

annotate + filter の組み合わせ #

コメント 5 件以上の記事 
Post.objects.annotate(
    comment_count=Count("comments")
).filter(comment_count__gte=5)

annotate で作った仮想フィールドを その場で filter 条件として 使えます。SQL の HAVING に変換されます。

filter + annotate の順序の落とし穴 #

🚫 意図と異なる結果 
# 公開されたコメントだけ数えたかったが…
Post.objects.filter(comments__published=True).annotate(
    cnt=Count("comments")
)
# JOIN 後に cnt が膨らむ可能性あり (重複行)
✅ Q と一緒に 
from django.db.models import Q

Post.objects.annotate(
    cnt=Count("comments", filter=Q(comments__published=True))
)

Count("comments", filter=Q(...)) が 集計時点で条件 をかける標準形です。

F 式 — DB 側の演算 #

Python で値を取得して計算し、再保存するパターンは レースコンディション を生みます。

🚫 レースコンディションの可能性 
post = Post.objects.get(pk=1)
post.views += 1
post.save()
# 2 つのリクエストが同時に読み書きすると、1 回が消える

F オブジェクトで DB 側で演算 します。

✅ F 式 
from django.db.models import F

Post.objects.filter(pk=1).update(views=F("views") + 1)
# UPDATE post SET views = views + 1 WHERE id = 1;

DB が原子的に処理するので並行性の問題がありません。他の活用例:

価格 10% 値上げ、2 フィールドの比較 
Product.objects.update(price=F("price") * Decimal("1.1"))

# 在庫が注文数量より少ない項目
Order.objects.filter(quantity__gt=F("product__stock"))

# 2 つのフィールドを同時更新
Account.objects.filter(pk=1).update(
    balance=F("balance") - 100,
    debit_count=F("debit_count") + 1,
)

Q オブジェクト — 複合条件 #

filter(...) のキーワード引数はすべて AND で結合されます。OR / NOT / 複合条件 が必要なら Q オブジェクト。

OR — タイトルまたは本文にキーワード 
from django.db.models import Q

Post.objects.filter(
    Q(title__icontains="Django") | Q(body__icontains="Django")
)
AND、NOT 
# AND
Post.objects.filter(Q(published=True) & Q(views__gte=100))

# NOT
Post.objects.filter(~Q(category__name="お知らせ"))

# 組み合わせ
Post.objects.filter(
    Q(published=True) & (Q(featured=True) | Q(views__gte=1000))
)

動的な条件の組み立て #

検索条件が動的なときに Q が輝きます。

検索フォーム処理 
def search_posts(query, only_published=True, category=None):
    q = Q()
    if query:
        q &= Q(title__icontains=query) | Q(body__icontains=query)
    if only_published:
        q &= Q(published=True)
    if category:
        q &= Q(category=category)
    return Post.objects.filter(q)

空の Q() から始めて条件を積み重ねるパターンです。CBV の get_queryset の中によく合います。

N+1 問題 — 最もありふれた性能の落とし穴 #

🚫 N+1 — コメントごとに記事の取得クエリが別々 
comments = Comment.objects.all()
for c in comments:
    print(c.post.title)        # 反復ごとに SELECT 1 回追加

クエリは最初の 1 回 (コメント全体) + N 回 (各コメントごとに記事取得)。コメント 1000 件なら 1001 クエリ。これが N+1 問題 です。

select_related — JOIN で 1 度に (FK 順方向) #

✅ select_related 
comments = Comment.objects.select_related("post").all()
for c in comments:
    print(c.post.title)        # 追加クエリ 0

生成される SQL:

SELECT comment.*, post.* FROM comment
LEFT OUTER JOIN post ON post.id = comment.post_id;

クエリ 1 回で JOIN により取得します。OneToOne、ForeignKey (順方向) で使えます。

prefetch_related — 別クエリで 1 度に (逆方向、M2M) #

逆方向 ForeignKey や ManyToMany は JOIN で 1 度に取得できません (行数が爆発する)。prefetch_related が 別クエリ + Python 側のマッピングで解決します。

🚫 N+1 — 記事ごとにコメント一覧 
posts = Post.objects.all()
for p in posts:
    print(len(p.comments.all()))   # 反復ごとにコメント取得
✅ prefetch_related 
posts = Post.objects.prefetch_related("comments").all()
for p in posts:
    print(len(p.comments.all()))   # 追加クエリ 0

生成されるクエリ: SELECT * FROM post + SELECT * FROM comment WHERE post_id IN (...) — 常に 2 クエリ。記事が 1000 件でもそのまま 2 回。

2 つを混ぜる #

両方一緒に 
posts = (Post.objects
         .select_related("author", "category")     # FK
         .prefetch_related("comments", "tags")     # 逆方向 / M2M
         .all())

Prefetch — prefetch のクエリ自体を加工 #

公開済みコメントだけ事前取得 
from django.db.models import Prefetch

posts = Post.objects.prefetch_related(
    Prefetch(
        "comments",
        queryset=Comment.objects.filter(published=True).order_by("-created_at"),
        to_attr="published_comments",
    )
)
for p in posts:
    for c in p.published_comments:
        print(c.body)

Prefetch オブジェクトで prefetch されるクエリ自体にフィルタ / 並び順 をかけられます。to_attr で別属性に格納し、元のマネージャとの衝突も避けます。

詳しい N+1 診断方法 (django-debug-toolbar、EXPLAIN、クエリ数アサーション) は 上級 #3 クエリ最適化 で扱います。

only / defer — 取得するフィールドの選択 #

大きなテキストフィールド (本文など) があり、一覧ページではあえて取りたくないとき。

only — 指定したフィールドのみ 
Post.objects.only("id", "title", "slug").all()
# 本文なしで軽いクエリ

Post.objects.defer("body").all()
# defer は逆 — 指定したフィールドを除く

注意: only したオブジェクトで 指定していないフィールドにアクセスすると自動で追加クエリ が発生します。意図せず N+1 になる可能性があります。

values / values_list — dict / tuple で受け取る #

モデルインスタンスではなく dict または tuple で受け取りたいとき。

values 
list(Post.objects.values("id", "title")[:3])
# [{'id': 1, 'title': '...'}, {'id': 2, 'title': '...'}, ...]

list(Post.objects.values_list("id", "title")[:3])
# [(1, '...'), (2, '...'), ...]

list(Post.objects.values_list("title", flat=True)[:3])
# ['...', '...', '...']    # flat=True なら単一値のリスト

API レスポンスを軽い dict にするとき、または id だけ集めて他のクエリに使うときに有用です。モデルインスタンスを作らないので メモリも少なく済みます。

サブクエリ — Subquery、OuterRef、Exists #

複雑な条件はサブクエリが答えです。

各記事の最新コメント作成日 
from django.db.models import Subquery, OuterRef

latest_comment = Comment.objects.filter(
    post=OuterRef("pk")
).order_by("-created_at")

posts = Post.objects.annotate(
    latest_comment_at=Subquery(latest_comment.values("created_at")[:1])
)
コメントがある記事だけ 
from django.db.models import Exists, OuterRef

has_comment = Comment.objects.filter(post=OuterRef("pk"))

Post.objects.annotate(
    has_comment=Exists(has_comment)
).filter(has_comment=True)

OuterRef("pk") が 外側クエリのカラムを参照 する印です。

最後のカード — raw SQL #

ORM での表現があまりにも複雑になったら raw SQL の出口があります。

raw — 結果をモデルインスタンスに 
posts = Post.objects.raw(
    "SELECT * FROM blog_post WHERE views > %s",
    [100],
)
for p in posts:
    print(p.title)
connection.cursor() — 完全な自由 
from django.db import connection

with connection.cursor() as cur:
    cur.execute(
        "SELECT category_id, COUNT(*) FROM blog_post GROUP BY category_id"
    )
    rows = cur.fetchall()

原則: ORM で解けるなら ORM。raw は本当に複雑か性能が決定的な場合だけ。パラメータは 必ず placeholder (%s) で、文字列フォーマット禁止 (SQL インジェクション防止)。

トランザクション — transaction.atomic #

複数の書き込みが 1 単位であるべきとき。

atomic ブロック 
from django.db import transaction

with transaction.atomic():
    order = Order.objects.create(user=user, total=total)
    for item in items:
        OrderItem.objects.create(order=order, product=item.product, qty=item.qty)
        Product.objects.filter(pk=item.product_id).update(
            stock=F("stock") - item.qty
        )

ブロック内で例外が出たら すべての書き込みがロールバック されます。デコレータ形式も可能 — @transaction.atomic で関数 / メソッド全体を 1 トランザクションに包む。トランザクション後処理 (on_commit) のような深いパターンは 上級 #5 で。

QuerySet の lazy な性質 #

QuerySet は 実際に評価されるまで SQL を送りません。

評価時点 
qs = Post.objects.filter(published=True)   # まだ SQL なし
qs = qs.order_by("-created_at")            # 依然なし
qs = qs.annotate(cnt=Count("comments"))    # 依然なし

list(qs)                                    # ここで SQL 実行
for p in qs: pass                           # またはここ
print(qs[0])                                # またはここ

チェーンしている間はただ クエリを組み立てる だけ。評価は反復 / スライシング / len / bool / list などで起きます。この性質を理解すれば 不要なクエリ を避けられます。

まとめ #

今回押さえたもの:

  • aggregate (全体集計 → dict)、annotate (行ごと集計 → QuerySet)
  • Count(..., filter=Q(...)) — 条件付き集計
  • F 式 — DB 側演算、並行性安全
  • Q オブジェクト — OR/AND/NOT、動的条件の組み立て
  • N+1 問題 とその解決: select_related (FK 順方向、JOIN)、prefetch_related (逆方向 / M2M、別クエリ)
  • Prefetch(..., queryset=..., to_attr=...) で prefetch 自体を加工
  • only / defer — フィールド選択 (ただし未選択フィールドアクセス時の追加クエリに注意)
  • values / values_list(flat=True) — dict/tuple で受け取る
  • Subquery、OuterRef、Exists — サブクエリ
  • raw SQL は最後のカード、パラメータは placeholder
  • transaction.atomic で複数の書き込みを 1 単位に
  • QuerySet は lazy — 評価時点の理解

次回 (#3 Signals と Middleware) では、モデル・ビューのフロー外で起きる 2 つ — Signals (イベント) と Middleware (リクエスト/レスポンスパイプライン) を扱います。両方とも強力ですが落とし穴も多いツールです。

X