(九)QuerySet 的构造、执行与缓存
1. 先得到查询描述,再读取结果
Book.objects 是模型管理器。filter(owner=alice) 返回 QuerySet,描述要读取的记录;继续 order_by("pk") 添加排序。构造这些描述时通常尚未发出 SELECT,遍历、转成列表或读取结果时才执行。
get 取得恰好一个对象,没有记录抛出 DoesNotExist;get_object_or_404 把没有找到转为页面 404。create 创建并保存,get_or_create 交回 (对象, 是否新建);values_list("name", flat=True) 直接读取单列值。
本章演示需要真实表和采集书目,统一用下面的临时环境准备:
# 项目文件:demos/lab.py
from contextlib import contextmanager
import os
from pathlib import Path
import re
from tempfile import TemporaryDirectory
import django
from django.core.management import call_command
from django.db import connections
@contextmanager
def lab():
# 每个演示脚本创建临时实例,使用项目真实迁移和采集书目。
with TemporaryDirectory(prefix="django-demo-") as folder:
os.environ["READING_INSTANCE"] = folder
os.environ["DJANGO_SETTINGS_MODULE"] = "reading_site.settings"
django.setup()
call_command("migrate", verbosity=0)
source = (
Path(__file__).resolve().parent.parent / "reading_project/data/books.csv"
)
call_command("seed_shelf", csv=source, verbosity=0)
try:
yield
finally:
connections.close_all()
def token(client, path):
page = client.get(path)
# 读取 Django 实际生成的隐藏字段,不根据 Cookie 文本猜表单令牌。
return re.search(r'name="csrfmiddlewaretoken" value="([^"]+)"', page.text)[1]call_command 在 Python 中执行已经认识的管理命令。lab 使用同一组迁移和真实 CSV,结束时关闭数据库连接;每个演示脚本独立启动一次环境。token 读取实际隐藏字段,给后面的登录和表单演示复用。
2. 一份查询缓存只对应自己的结果
# 项目文件:demos/query_steps.py
from lab import lab
with lab():
from django.contrib.auth import get_user_model
from django.db import connection
from django.test.utils import CaptureQueriesContext
from library.models import Book
alice = get_user_model().objects.get(username="alice")
# 分开统计构造、首次遍历和再次遍历,观察查询缓存。
with CaptureQueriesContext(connection) as built:
books = Book.objects.filter(owner=alice).order_by("pk")
with CaptureQueriesContext(connection) as first:
titles = [book.title for book in books]
with CaptureQueriesContext(connection) as cached:
again = [book.title for book in books]
print("构造、首次遍历、再次遍历", len(built), len(first), len(cached))
print("结果相同", titles == again, "条数", len(titles))
with CaptureQueriesContext(connection) as naive:
# 每本书分别读取 owner 和 tags,会不断触发关联查询。
a = [
(b.title, b.owner.username, [t.name for t in b.tags.all()])
for b in Book.objects.filter(owner=alice)
]
with CaptureQueriesContext(connection) as optimized:
# 外键随主查询连接,多对多关系再批量读取一次。
b = [
(b.title, b.owner.username, [t.name for t in b.tags.all()])
for b in Book.objects.filter(owner=alice)
.select_related("owner")
.prefetch_related("tags")
]
print("逐本查询", len(naive), "关联加载", len(optimized), "结果相同", a == b)构造、首次遍历、再次遍历分别得到查询数 0、1、0。首次遍历把结果保存在这份 QuerySet 的缓存中,再遍历复用;追加筛选产生新的 QuerySet,仍需要执行新的查询。
count() 适合直接问数量,exists() 适合只问是否存在。如果接下来还要完整遍历,同一个任务里先 exists() 再遍历可能产生两次读取,先确定实际需要哪些结果,再决定调用方式。