Python 现代开发学习笔记

(九)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() 再遍历可能产生两次读取,先确定实际需要哪些结果,再决定调用方式。

构造 filter 与排序时只形成描述,首次遍历执行 SQL 并缓存六本,再次遍历复用,添加新筛选后形成另一份待执行查询

笔记正文与全部题解 · Ctrl / ⌘ K