Django4：ORM——查询、更新、统计与事务

1. QuerySet：一个还没执行的查询

from blog.models import Category, Post, Tag

posts = Post.objects.filter(status=Post.Status.PUBLISHED)

Post 是模型，objects 是管理器，filter() 返回一个 QuerySet。可以把它理解成“一条可以继续加条件的查询”，而不是已经装进内存的列表。

QuerySet 是惰性的：只有在遍历、list()、切片取值、len() 或打印时才真正访问数据库。所以可以放心地一层层组合条件。

2. 增：创建记录和关系

from django.contrib.auth import get_user_model

author = get_user_model().objects.get(username="admin")
category, _ = Category.objects.get_or_create(name="技术")
tag, _ = Tag.objects.get_or_create(name="Python")

post = Post(
    title="使用 ORM 创建一篇文章",
    slug="orm-practice-post",
    body="这篇文章专门用于查询练习。",
    author=author,
    category=category,
)
post.full_clean()
post.save()
post.tags.add(tag)

三步分别是：full_clean() 验证字段 → save() 写入文章表 → tags.add() 写入标签中间表。

3. 查：一条、多条、不存在

post = Post.objects.get(slug="first-django-post")
posts = Post.objects.filter(author=author, status=Post.Status.PUBLISHED)
latest = posts.first()
has_posts = posts.exists()
total = posts.count()

在视图里查单个对象，常用 get_object_or_404()，它把“没找到”直接变成 404 页面（第 5 篇）。

4. 条件、跨关系查询与排序

from django.db.models import Q
from django.utils import timezone
from datetime import timedelta

recent = Post.objects.filter(created_at__gte=timezone.now() - timedelta(days=7))
matches = Post.objects.filter(
    Q(title__icontains="Django") | Q(body__icontains="Django"),
    status=Post.Status.PUBLISHED,
)
by_author = Post.objects.filter(author__username="admin")
by_tag = Post.objects.filter(tags__name__in=["Django", "Python"]).distinct()
latest_ten = Post.objects.filter(status="published").order_by("-created_at", "-pk")[:10]

双下划线 有两种用法：

• created_at__gte：字段 + 比较方式（大于等于）。常用的还有 exact、icontains、in、lt。
• author__username：跨关系访问，相当于 SQL 的 JOIN。

条件组合： 多个参数默认是“且”；“或”要用 Q(...) | Q(...)。

去重： 一篇文章同时有 Django 和 Python 两个标签时，按标签筛选会返回两行，所以加 .distinct()。

5. 只取需要的字段，以及分组统计

from django.db.models import Count

rows = list(Post.objects.values("id", "title", "status")[:5])
titles = list(Post.objects.values_list("title", flat=True)[:5])

categories = Category.objects.annotate(
    published_count=Count("posts", filter=Q(posts__status=Post.Status.PUBLISHED))
).order_by("-published_count", "name")

for category in categories:
    print(category.name, category.published_count)

• values() 返回字典，values_list(..., flat=True) 返回单值列表。它们比完整对象轻，但不能再调用模型方法。
• annotate() 给每一行加一个统计列（这里是每个分类的已发布文章数）；aggregate() 则对整组数据算出一个汇总值。
• filter=Q(...) 只统计已发布的文章，同时保留了一篇公开文章都没有的分类（数量为 0）。

6. 改：save() 和 update() 的区别

post = Post.objects.get(slug="orm-practice-post")
post.title = "使用 ORM 修改一篇文章"
post.save(update_fields=["title", "updated_at"])

changed = Post.objects.filter(pk=post.pk, status=Post.Status.DRAFT).update(
    status=Post.Status.PUBLISHED,
    updated_at=timezone.now(),
)
print(changed)

save(update_fields=[...]) 只写列出的字段；想让 updated_at 也更新，就要把它写进列表。

changed 的值有用：条件里写了“仍是草稿”，如果返回 0，说明这篇文章已经不是草稿了。

7. F 表达式：在数据库里做加法

from django.db.models import F

Post.objects.filter(pk=post.pk).update(views=F("views") + 1)
post.refresh_from_db()
print(post.views)

如果两个请求同时“读出 10 → 各自加 1 → 都保存成 11”，就丢了一次计数。F("views") + 1 把加法交给数据库在一条语句里完成，基于当前值递增，不会丢。

内存里的 post 对象不会自动变，用 refresh_from_db() 重新读取。

8. 事务：要么全成功，要么全回滚

需求：修改文章标题和标签，必须同时成功。

from django.db import transaction

with transaction.atomic():
    post = Post.objects.get(slug="orm-practice-post")
    post.title = "事务中修改的文章标题"
    post.save(update_fields=["title", "updated_at"])
    post.tags.set([tag])

atomic() 块里任何一步抛出异常，之前的数据库写入都会撤销；正常结束才提交。

捕获数据库错误时，把 try/except 放在 atomic() 块的外面。

9. N+1 问题：模板里一个循环，数据库被查了几十次

一页显示 10 篇文章，每篇都要显示作者、分类和标签。如果直接写：

posts = Post.objects.filter(status=Post.Status.PUBLISHED)

模板循环里每访问一次 post.author、post.category、post.tags.all，都会再查一次数据库。10 篇文章就是 1 + 10 × 3 次查询——这就是 N+1 问题。

在下面的演示里拖动文章数，再勾选两个优化选项，看查询次数的变化：

解决方法是在查询时告诉 Django 你会用到哪些关系：

posts = (
    Post.objects.filter(status=Post.Status.PUBLISHED)
    .select_related("author", "category")
    .prefetch_related("tags")
)

优化后，分页列表固定是 3 次查询（总数、文章 + 作者 + 分类、标签），与文章数量无关。第 8 篇有一个测试专门检查这个数字。

10. 删：先确认范围

practice = Post.objects.get(slug="orm-practice-post")
print(practice.pk, practice.title)
practice.delete()

这会删除文章和它的标签关系，但标签本身还在。

小练习

自己先写，再展开对照。

list(Post.objects.filter(
    status=Post.Status.PUBLISHED,
    created_at__gte=timezone.now() - timedelta(days=7),
).values_list("title", flat=True))

from django.contrib.auth import get_user_model
from django.db.models import Count, Q

get_user_model().objects.annotate(
    published=Count("posts", filter=Q(posts__status="published"))
).values("username", "published")