编程 Django 6.1 深度实战：当二十年老牌框架学会「按需加载」——从 Fetch Mode 到数据库级级联删除、多Mailer 架构与生产级迁移的完全指南（2026）

2026-06-18 07:56:48 +0800 CST views 45

Django 6.1 深度实战：当二十年老牌框架学会「按需加载」——从 Fetch Mode 到数据库级级联删除、多Mailer 架构与生产级迁移的完全指南（2026）

引言：Django 的又一次进化

2026 年 8 月，Django 6.1 正式版即将发布。对于这个已经走过二十年历程的 Python Web 框架来说，每一次大版本更新都像是一场精心策划的手术——既要切除长期积累的病灶，又要保证老病人平稳过渡。

Django 6.1 这次手术的焦点，恰好落在了一个所有 Django 开发者都痛过、但一直没找到根治方案的领域：ORM 的查询效率。

N+1 查询问题是 Django ORM 的「慢性病」。你知道它存在，你用 select_related() 和 prefetch_related() 去治它，但你永远无法确认自己是不是漏了某个字段、某个场景。select_related() 漏了？API 响应慢三秒。prefetch_related() 多了？白白加载一大堆不需要的数据。这是一个需要你时刻提防、却永远无法彻底消除的隐患。

Django 6.1 用 Fetch Mode 给出了答案。这不是一个权宜之计，而是 ORM 查询语义的一次范式级重构。配合数据库级的 on_delete 和多 Mailer 架构，6.1 版本让 Django 在性能和架构层面都迈出了重要一步。

本文将从这三个核心特性出发，深入剖析其设计哲学、实现原理、实战用法，以及生产级迁移中需要注意的每一个坑点。

一、Fetch Mode：ORM 查询的「自动驾驶」模式

1.1 N+1 问题的本质

在理解 Fetch Mode 之前，我们需要彻底搞清楚 N+1 问题的本质。这不是一个简单的「循环里查数据库」的问题。

# 经典的 N+1 场景
books = Book.objects.all()  # 1 次查询：获取所有 book
for book in books:
    print(book.author.name)  # N 次查询：每本书都查一次 author

问题在于：当你执行 Book.objects.all() 时，Django 只加载了 Book 表的字段。author 是一个 ForeignKey，它只加载了 author_id（一个整数）。当你访问 book.author 时，Django 发现它还没有加载完整的 Author 对象，于是发起一次额外的查询。

这是一种延迟加载（Lazy Loading）策略。它的设计初衷是好的：如果你不需要 author，就不加载它，节省内存和网络开销。但现实是，延迟加载在批量场景下变成了灾难——你永远无法知道一次「看似简单」的属性访问会触发多少次数据库查询。

1.2 传统解决方案的痛点

# select_related：JOIN 模式，适合 ForeignKey 和 OneToOne
books = Book.objects.select_related('author')  # 1 次查询，JOIN author 表
for book in books:
    print(book.author.name)  # 0 次额外查询

# prefetch_related：独立查询模式，适合 ManyToMany 和反查
authors = Author.objects.prefetch_related('books')  # 2 次查询
for author in authors:
    for book in author.books.all():
        print(book.title)

这两个工具是 Django 开发者的日常武器，但它们有三个根本性的痛点：

痛点一：你必须提前知道要访问哪些字段

这不是一个技术问题，而是一个认知问题。在复杂业务逻辑中，你调用 Book.objects.all() 时，可能还不确定后续代码会访问哪些关联字段。 serializer 层可能要 author，模板层可能要 author.publisher，中间件可能要 author.avatar……你得像算命一样提前预判所有可能的访问路径。

# 你得写这么长的一串
books = Book.objects.select_related(
    'author', 'author__publisher', 'author__publisher__country'
).prefetch_related('tags', 'reviews')

痛点二：过度预加载浪费资源

为了保险，你可能会加载所有可能用到的关联字段。但 API 端点 A 只需要 author.name，端点 B 需要 author.publisher，你却为所有端点加载了所有关联数据。这就像为了偶尔下雨，每天出门都带伞箱——过于保守。

痛点三：维护成本高

代码迭代时，新增了一个字段访问，忘记更新 select_related，性能就悄悄恶化。没有编译器帮你检查，没有 linter 帮你警告。这是一个「静默故障」——不会报错，只是越来越慢。

1.3 Fetch Mode 的设计哲学

Django 6.1 的 Fetch Mode 不是在 select_related / prefetch_related 之上又加一层工具，而是重新定义了 ORM 的查询语义：

当你访问一个未加载的字段时，Django 应该做什么？

之前这个问题的答案是唯一的：发起一次查询，加载当前实例的该字段。Fetch Mode 让你可以选择不同的答案：

FETCH_ONE：「老规矩，查一个」
FETCH_PEERS：「既然要查，把兄弟姐妹一起查了吧」
RAISE：「不许查，报错！」

这三种模式的语义非常清晰：

模式	行为	等价于	适用场景
FETCH_ONE	为当前实例查一次	默认行为	精确控制、单实例访问
FETCH_PEERS	为整个 QuerySet 批量查	自动 prefetch_related	批量遍历、列表渲染
RAISE	抛异常	禁止延迟加载	性能关键路径、API 端点

FETCH_PEERS 是杀手级特性。它本质上是 Django 之前一直缺失的「自动 prefetch」——你不需要提前声明要预加载哪些字段，Django 在你首次访问时自动批量加载整个 QuerySet 对应的关联数据。

1.4 Fetch Mode 的内部实现原理

Fetch Mode 的实现依赖于 Django ORM 内部一个关键机制：QuerySet 的结果缓存共享。

当你执行 Book.objects.all() 时，Django 返回一个 QuerySet。这个 QuerySet 在被迭代时，会缓存所有 Book 实例。这些实例共享同一个「出身信息」——它们知道自己来自哪个 QuerySet。

# Django 内部大致实现（简化）
class QuerySet:
    def fetch_mode(self, mode):
        # 设置 fetch_mode，返回一个新的 QuerySet
        qs = self._clone()
        qs._fetch_mode = mode
        return qs

class ModelInstance:
    def __getattr__(self, name):
        field = self._meta.get_field(name)
        if field.is_relation and not self._is_loaded(name):
            mode = self._fetch_mode  # 从 QuerySet 继承的 mode
            
            if mode == FETCH_ONE:
                # 只查当前实例的关联对象
                self._fetch_one(field)
            elif mode == FETCH_PEERS:
                # 查整个 QuerySet 的所有实例的关联对象
                self._fetch_peers(field)
            elif mode == RAISE:
                raise FieldFetchBlocked(
                    f"Field '{name}' is blocked from on-demand fetching"
                )

FETCH_PEERS 的核心逻辑：

当前实例首次访问 book.author 时，Django 发现 author 未加载
Django 查找当前实例所属的 QuerySet 缓存
收集该 QuerySet 中所有实例的 author_id 值
执行一次批量查询：Author.objects.filter(id__in=[所有author_id])
将查询结果填充到所有实例的 author 缓存中
后续实例访问 author 时，直接从缓存读取，零查询

关键细节：FETCH_PEERS 只在首次访问时触发批量查询。这意味着：

第一本书访问 author → 1 次批量查询
后续所有书访问 author → 0 次查询
总查询数 = 1（books） + 1（authors） = 2

这和手写 select_related('author') 的效果完全一样，但你不需要提前声明。

1.5 三种模式实战代码

FETCH_ONE（默认模式）

from django.db import models

# FETCH_ONE 是默认行为，不需要显式设置
books = Book.objects.all()  # 等价于 .fetch_mode(models.FETCH_ONE)
for book in books:
    # 每次访问 author 都触发 1 次查询
    print(book.author.name)  # N+1！

什么时候用 FETCH_ONE？

你只处理单个实例（不是批量遍历）
你需要精确控制每个查询的时机
你在写一个只查一条记录的 API 端点

# 单实例场景：FETCH_ONE 是合理的
book = Book.objects.get(pk=1)
print(book.author.name)  # 1 次额外查询，没问题

FETCH_PEERS（批量优化模式）

from django.db import models

books = Book.objects.fetch_mode(models.FETCH_PEERS)
for book in books:
    print(book.author.name)  # 只有 2 次查询！

# 多层关联也能自动优化
books = Book.objects.fetch_mode(models.FETCH_PEERS)
for book in books:
    # 第一次访问 author → 批量加载所有 author
    # 第一次访问 author.publisher → 批量加载所有 publisher
    print(book.author.name, book.author.publisher.company_name)
    # 总查询：1(books) + 1(authors) + 1(publishers) = 3

什么时候用 FETCH_PEERS？

批量遍历场景（列表页、批量处理任务）
API 端点返回序列化列表
你不确定后续代码会访问哪些关联字段
你想消除 N+1 但不想维护 prefetch_related 列表

# DRF Serializer 列表场景
class BookListView(APIView):
    def get(self, request):
        books = Book.objects.fetch_mode(models.FETCH_PEERS)
        serializer = BookSerializer(books, many=True)
        return Response(serializer.data)
        # Serializer 访问 book.author、book.tags 等时，
        # FETCH_PEERS 自动批量加载，无需在 View 里写 prefetch_related

FETCH_PEERS 和 select_related / prefetch_related 可以共存：

# 你可以混合使用：已知的关联用 select_related 预加载，
# 未知的关联让 FETCH_PEERS 按需处理
books = Book.objects.select_related('author').fetch_mode(models.FETCH_PEERS)
for book in books:
    print(book.author.name)       # 0 次额外查询（已 select_related）
    print(book.author.publisher)  # FETCH_PEERS 自动批量加载
    print(book.tags.all())        # FETCH_PEERS 自动批量加载

RAISE（严格模式）

from django.db import models

# RAISE 模式：禁止任何延迟加载
books = Book.objects.select_related('author').fetch_mode(models.RAISE)
for book in books:
    print(book.title)       # ✅ 已加载的字段，正常访问
    print(book.author.name) # ✅ 已 select_related，正常访问
    print(book.tags.all())  # ❌ FieldFetchBlocked！tags 未预加载

什么时候用 RAISE？

性能关键路径（低延迟 API、高频接口）
你想确保没有「偷跑」的查询
CI/CD 中作为性能测试的断言工具
你想强制团队在写代码时明确声明所有关联字段

# API 性能断言：确保这个端点没有额外查询
def test_api_no_lazy_queries():
    with self.assertNumQueries(2):  # 期望只有 2 次查询
        books = Book.objects.select_related('author').fetch_mode(models.RAISE)
        for book in books:
            book.title
            book.author.name
            # 如果这里访问了 book.publisher，RAISE 会报错
            # CI 就会失败，提醒你补上 select_related('publisher')

RAISE 模式是 Django 开发者一直想要但从未拥有的性能护栏。它把「隐式查询」变成了「显式错误」，让你在开发阶段就能发现 N+1 问题，而不是在生产环境的慢日志里才看到。

1.6 Fetch Mode 与 QuerySet 链式调用的交互

Fetch Mode 是一个 QuerySet 方法，支持链式调用：

# 链式调用
books = (
    Book.objects
    .filter(status='published')
    .order_by('-created_at')
    .fetch_mode(models.FETCH_PEERS)
    .only('title', 'author')  # 只加载 title 和 author_id
)

for book in books:
    print(book.title)
    print(book.author.name)   # FETCH_PEERS 批量加载
    print(book.description)   # FETCH_PEERS 批量加载 description

Fetch Mode 和 only() / defer() 的交互：

# defer + FETCH_PEERS
books = Book.objects.defer('description').fetch_mode(models.FETCH_PEERS)
for book in books:
    print(book.title)           # 已加载
    print(book.description)     # 被 defer 了，但 FETCH_PEERS 会批量加载
    # 这意味着 defer + FETCH_PEERS 的组合在首次访问 description 时
    # 会触发一次批量加载所有实例的 description

注意：defer() 和 only() 的语义在 FETCH_PEERS 下发生了微妙的变化。defer() 的初衷是「我不需要这个字段，别加载它」，但 FETCH_PEERS 会在首次访问时批量加载。如果你真的不想某个字段被加载，应该用 RAISE 模式来保护：

books = (
    Book.objects
    .defer('description')
    .fetch_mode(models.RAISE)  # 描述被 defer 且禁止按需加载
)
for book in books:
    print(book.title)           # ✅
    print(book.description)     # ❌ FieldFetchBlocked！你必须显式处理

1.7 Fetch Mode 的性能基准测试

让我们用数据说话。以下是在一个包含 1000 本书、500 个作者的数据库上的基准测试：

import time
from django.db import models, connection
from django.test.utils import override_settings

def benchmark_fetch_modes():
    """基准测试：三种 Fetch Mode 的查询数和耗时"""
    
    # 强制重置查询计数
    from django.db import reset_queries
    
    # 测试 FETCH_ONE（传统 N+1）
    reset_queries()
    start = time.time()
    books = Book.objects.fetch_mode(models.FETCH_ONE)[:100]
    for book in books:
        _ = book.author.name
    fetch_one_queries = len(connection.queries)
    fetch_one_time = time.time() - start
    
    # 测试 FETCH_PEERS
    reset_queries()
    start = time.time()
    books = Book.objects.fetch_mode(models.FETCH_PEERS)[:100]
    for book in books:
        _ = book.author.name
    fetch_peers_queries = len(connection.queries)
    fetch_peers_time = time.time() - start
    
    # 测试 select_related（传统手动优化）
    reset_queries()
    start = time.time()
    books = Book.objects.select_related('author')[:100]
    for book in books:
        _ = book.author.name
    select_related_queries = len(connection.queries)
    select_related_time = time.time() - start
    
    print(f"FETCH_ONE:     {fetch_one_queries} queries, {fetch_one_time:.3f}s")
    print(f"FETCH_PEERS:   {fetch_peers_queries} queries, {fetch_peers_time:.3f}s")
    print(f"select_related: {select_related_queries} queries, {select_related_time:.3f}s")

# 预期结果：
# FETCH_ONE:      101 queries, 0.850s  (1 + 100)
# FETCH_PEERS:    2 queries,   0.012s  (1 + 1)
# select_related: 1 queries,   0.008s  (JOIN)

FETCH_PEERS 和 select_related 的查询次数接近（2 vs 1），但 select_related 用 JOIN 方式更高效。FETCH_PEERS 用两次独立查询，在关联数据量大时可能比 JOIN 更好（避免 JOIN 结果集膨胀）。

关键洞察：FETCH_PEERS 不是 select_related 的替代品，而是 prefetch_related 的自动化版本。它的优势在于你不需要提前知道要预加载什么。

1.8 Fetch Mode 的局限性和注意事项

局限性一：不适用于切片后的 QuerySet

books = Book.objects.fetch_mode(models.FETCH_PEERS)[:10]
for book in books:
    print(book.author.name)
# ⚠️ 切片后的 QuerySet 已经执行了查询，
# FETCH_PEERS 可能无法追踪所有 peer 实例

局限性二：跨 QuerySet 实例无法共享

# 这两个 QuerySet 是独立的
books_a = Book.objects.filter(year=2025).fetch_mode(models.FETCH_PEERS)
books_b = Book.objects.filter(year=2026).fetch_mode(models.FETCH_PEERS)
# books_a 中访问 author 会批量加载 2025 年的 author
# books_b 中访问 author 会批量加载 2026 年的 author
# 它们不会合并成一次查询

局限性三：只对「同源」实例有效

books = Book.objects.fetch_mode(models.FETCH_PEERS)
# 如果你在循环中创建了新的 Book 实例（非数据库查询结果），
# 这个新实例不在 QuerySet 缓存中，FETCH_PEERS 不会覆盖它
manual_book = Book(title="手动创建", author_id=1)
# manual_book 不属于任何 QuerySet，访问 author 时走 FETCH_ONE

二、数据库级 on_delete：让数据库干它该干的事

2.1 为什么需要数据库级的 on_delete

Django 的 on_delete 参数从框架诞生之初就存在，但它的所有选项（CASCADE、PROTECT、SET_NULL、SET_DEFAULT、DO_NOTHING）都是在 Python 层面 实现的。

这意味着什么？当你删除一个 Author 时，Django 的 CASCADE 流程是：

查询所有关联的 Book 对象（SELECT）
对每个 Book 对象触发 pre_delete 信号
执行 DELETE SQL 删除所有 Book
对每个 Book 对象触发 post_delete 信号

对于一个有 10 万本书的作者，这个过程会：

加载 10 万个 Book 实例到内存
发送 10 万次 pre_delete 和 post_delete 信号
执行 10 万次 DELETE（或者 Django 优化后的批量 DELETE）

而数据库级的 CASCADE 只需要：

一条 SQL：DELETE FROM book WHERE author_id = ?（数据库内部自动级联）
或者数据库在 Foreign Key 定义时就声明了 ON DELETE CASCADE，连这条 DELETE 都不需要——数据库自动处理

效率差距是巨大的：Python 级 CASCADE 需要加载和遍历所有对象，数据库级 CASCADE 是一条 SQL 或零 SQL。

2.2 三个新的数据库级选项

Django 6.1 新增了三个数据库级 on_delete 选项：

from django.db.models import DB_CASCADE, DB_SET_NULL, DB_SET_DEFAULT

class Book(models.Model):
    author = models.ForeignKey(
        'Author',
        on_delete=DB_CASCADE,  # 数据库级级联删除
    )
    
    category = models.ForeignKey(
        'Category',
        on_delete=DB_SET_NULL,  # 数据库级置空
        null=True,
    )
    
    publisher = models.ForeignKey(
        'Publisher',
        on_delete=DB_SET_DEFAULT,  # 数据库级设为默认值
        default=1,  # 必须有 default 值
    )

这三个选项在数据库层面生成的 SQL：

-- DB_CASCADE
ALTER TABLE book 
ADD CONSTRAINT fk_book_author 
FOREIGN KEY (author_id) REFERENCES author(id) 
ON DELETE CASCADE;

-- DB_SET_NULL
ALTER TABLE book 
ADD CONSTRAINT fk_book_category 
FOREIGN KEY (category_id) REFERENCES category(id) 
ON DELETE SET NULL;

-- DB_SET_DEFAULT
ALTER TABLE book 
ADD CONSTRAINT fk_book_publisher 
FOREIGN KEY (publisher_id) REFERENCES publisher(id) 
ON DELETE SET DEFAULT;

2.3 DB_CASCADE vs CASCADE：深度对比

# Python 级 CASCADE（Django 传统行为）
from django.db.models import CASCADE

class Book(models.Model):
    author = models.ForeignKey('Author', on_delete=CASCADE)

# 删除一个有 100000 本书 的作者
author = Author.objects.get(pk=1)
author.delete()
# Django 的执行流程：
# 1. SELECT * FROM book WHERE author_id = 1  → 加载 100000 个 Book 实例
# 2. 对每个 Book 发 pre_delete 信号
# 3. DELETE FROM book WHERE author_id = 1
# 4. 对每个 Book 发 post_delete 信号
# 5. DELETE FROM author WHERE id = 1
# 耗时：取决于 100000 个实例的加载和信号处理

# 数据库级 DB_CASCADE
from django.db.models import DB_CASCADE

class Book(models.Model):
    author = models.ForeignKey('Author', on_delete=DB_CASCADE)

# 删除同一个作者
author = Author.objects.get(pk=1)
author.delete()
# Django 的执行流程：
# 1. DELETE FROM author WHERE id = 1
# 数据库自动级联删除所有关联的 book
# 耗时：一条 DELETE + 数据库内部的级联操作

核心差异：

维度	CASCADE (Python)	DB_CASCADE (数据库)
加载关联对象	是（SELECT + 内存）	否
信号触发	pre_delete + post_delete	无
SQL 数量	1 SELECT + 1 DELETE (books) + 1 DELETE (author)	1 DELETE (author)
内存消耗	O(N)（N = 关联对象数）	O(1)
速度	慢（加载+信号+删除）	快（数据库内部级联）
数据库依赖	无（跨数据库通用）	需数据库支持 FK ON DELETE
事务一致性	Django 事务保证	数据库 FK 约束保证

2.4 DB_CASCADE 的信号问题

这是最重要的注意事项：DB_CASCADE 不触发 pre_delete 和 post_delete 信号。

这看起来是个缺陷，但其实是设计上的必然选择。数据库级联删除发生在数据库内部，Django 根本不知道哪些记录被删除了。信号的前提是「Django 知道一个对象即将被删除」，但 DB_CASCADE 让 Django 对关联对象的删除完全无感知。

# 信号不会触发
from django.db.models.signals import pre_delete, post_delete

def on_book_delete(sender, instance, **kwargs):
    print(f"Book {instance.title} is being deleted")
    # 清理缓存、通知其他系统、记录审计日志等

pre_delete.connect(on_book_delete, sender=Book)

# 使用 DB_CASCADE 时，删除 Author 不会触发 Book 的 pre_delete
# 上面的 on_book_delete 永远不会被调用！

什么时候必须用 CASCADE 而不能用 DB_CASCADE：

你依赖 pre_delete / post_delete 信号做清理工作
你需要在删除前做验证（如检查关联对象的状态）
你需要记录审计日志
你需要通知其他系统（如搜索索引、缓存）
你有自定义的删除逻辑

什么时候用 DB_CASCADE 是更好的选择：

纯数据级联，无业务逻辑依赖
大量关联数据，Python 级 CASCADE 性能不可接受
你不需要信号
批量清理数据的定时任务

# 混合策略：大部分用 DB_CASCADE，关键业务用 CASCADE
class LogEntry(models.Model):
    # 审计日志：级联删除时需要触发信号来通知监控系统
    user = models.ForeignKey('User', on_delete=CASCADE)  # Python 级

class PageView(models.Model):
    # 页面访问记录：纯数据，不需要信号，量大
    user = models.ForeignKey('User', on_delete=DB_CASCADE)  # 数据库级

2.5 DB_SET_NULL 和 DB_SET_DEFAULT 实战

# DB_SET_NULL：删除分类后，书籍的分类字段置空
class Book(models.Model):
    category = models.ForeignKey(
        'Category',
        on_delete=DB_SET_NULL,
        null=True,  # 必须！DB_SET_NULL 要求字段允许 NULL
    )

# DB_SET_NULL 不触发信号，但会自动将 category_id 设为 NULL
# 这对「软依赖」场景非常适合：分类没了，书籍还在，只是变成「未分类」

# DB_SET_DEFAULT：删除出版社后，书籍的出版社设为默认值
class Book(models.Model):
    publisher = models.ForeignKey(
        'Publisher',
        on_delete=DB_SET_DEFAULT,
        default=1,  # 必须！DB_SET_DEFAULT 要求字段有 default 值
        # default 可以是一个 PK 值或 callable
    )

# 注意：default=1 意味着数据库会执行 SET DEFAULT，将 publisher_id 设为 1
# 但这个 default=1 的 Publisher 必须存在于数据库中！否则 FK 约束会失败

DB_SET_DEFAULT 的陷阱：

# ❌ 错误用法：default 值在数据库中不存在
class Book(models.Model):
    publisher = models.ForeignKey(
        'Publisher',
        on_delete=DB_SET_DEFAULT,
        default=999,  # Publisher 999 可能不存在！
    )

# 当删除一个 Publisher 时，数据库会尝试将关联的 book.publisher_id 设为 999
# 如果 Publisher 999 不存在，FK 约束检查会失败，整个删除操作会回滚！

# ✅ 正确用法：确保 default 值存在
DEFAULT_PUBLISHER_ID = 1  # 在 migration 或 fixture 中确保这个 ID 存在

class Book(models.Model):
    publisher = models.ForeignKey(
        'Publisher',
        on_delete=DB_SET_DEFAULT,
        default=DEFAULT_PUBLISHER_ID,
    )

2.6 数据库级 on_delete 的 Migration 处理

当你从 CASCADE 迁移到 DB_CASCADE 时，需要生成新的 migration：

# migration: 0002_switch_to_db_cascade.py
from django.db import migrations, models
from django.db.models import DB_CASCADE

class Migration(migrations.Migration):
    dependencies = [('books', '0001_initial')]
    
    operations = [
        migrations.AlterField(
            model_name='book',
            name='author',
            field=models.ForeignKey(
                'Author',
                on_delete=DB_CASCADE,  # 从 CASCADE 改为 DB_CASCADE
            ),
        ),
    ]

Migration 执行时的关键步骤：

Django 会生成 ALTER TABLE 语句，修改 FK 约束
如果数据库已经有数据，ALTER TABLE 可能需要锁表
PostgreSQL 和 MySQL 在大表上 ALTER FK 可能很慢

# 在生产环境执行 migration 时的建议
# 1. 先在 staging 环境测试
# 2. 使用 --dry-run 查看会执行哪些 SQL
python manage.py migrate --dry-run

# 3. 大表场景：考虑手动执行 SQL，避免 Django migration 的锁表时间过长
# 先添加新的 FK 约束（不带 ON DELETE），然后删除旧约束，再修改新约束加上 ON DELETE
# 这种分步方式可以减少锁表时间

三、MAILERS：邮件架构的「DATABASES 化」重构

3.1 旧邮件架构的问题

Django 的邮件配置从诞生之日起就只有一个模式：

# settings.py - 旧方式
EMAIL_BACKEND = 'django.core.mail.backends.smtp.EmailBackend'
EMAIL_HOST = 'smtp.example.com'
EMAIL_PORT = 587
EMAIL_USE_TLS = True
EMAIL_HOST_USER = 'noreply@example.com'
EMAIL_HOST_PASSWORD = 'secret'

这套配置有三个问题：

问题一：只能配置一个邮件后端

你的应用可能需要多个邮件通道：

系统通知邮件用 SMTP 直连
营销邮件用第三方服务（如 SendGrid、Mailgun）
开发环境用文件后端（写到文件不真正发送）
测试环境用内存后端

但 EMAIL_BACKEND 只能设一个值。切换后端需要改代码或改配置，非常不灵活。

问题二：配置散乱

EMAIL_HOST、EMAIL_PORT、EMAIL_USE_TLS、EMAIL_HOST_USER、EMAIL_HOST_PASSWORD……这些都是全局变量，散落在 settings.py 里，和 DATABASES、CACHES 的整洁配置形成鲜明对比。

问题三：无法动态切换

在同一个请求中，你可能需要用不同后端发送不同类型的邮件。旧架构不支持这种场景。

3.2 MAILERS 的设计模式

Django 6.1 引入了 MAILERS 设置，完全遵循 DATABASES / CACHES / STORAGES 的配置模式：

# settings.py - 新方式
MAILERS = {
    "default": {
        "BACKEND": "django.core.mail.backends.smtp.EmailBackend",
        "OPTIONS": {
            "host": "smtp.example.com",
            "port": 587,
            "use_tls": True,
            "username": "noreply@example.com",
            "password": "secret",
        },
    },
    "marketing": {
        "BACKEND": "sendgrid_backend.SendgridBackend",
        "OPTIONS": {
            "api_key": "SG.xxx...",
        },
    },
    "transactional": {
        "BACKEND": "django.core.mail.backends.smtp.EmailBackend",
        "OPTIONS": {
            "host": "smtp.transactional.example.com",
            "port": 465,
            "use_ssl": True,
            "username": "transactional@example.com",
            "password": "another_secret",
        },
    },
    "dev": {
        "BACKEND": "django.core.mail.backends.filebased.EmailBackend",
        "OPTIONS": {
            "file_path": "/tmp/app-messages/",
        },
    },
}

架构模式对比：

特性	DATABASES	CACHES	STORAGES	MAILERS（新）
多实例	✅	✅	✅	✅
BACKEND 选择	✅	✅	✅	✅
OPTIONS 配置	✅	✅	✅	✅
using 参数	✅	✅	✅	✅
default 别名	✅	✅	✅	✅

3.3 MAILERS 的使用方式

from django.core import mail

# 使用 default mailer（无需指定）
mail.send_mail(
    "系统通知",
    "你的订单已发货",
    "noreply@example.com",
    ["user@example.com"],
)

# 使用 marketing mailer
mail.send_mail(
    "新品推荐",
    "查看我们最新的产品系列",
    "marketing@example.com",
    ["user@example.com"],
    using="marketing",  # 指定使用哪个 mailer
)

# 获取特定 mailer 的 backend 实例
marketing_mailer = mail.mailers["marketing"]
marketing_mailer.send([
    mail.EmailMessage(
        "营销邮件",
        "特别优惠...",
        "marketing@example.com",
        ["subscriber1@example.com", "subscriber2@example.com"],
    )
])

# 批量发送到不同 mailer
with mail.mailers["transactional"] as connection:
    mail.EmailMessage(
        "交易确认",
        "你的支付已成功",
        "transactional@example.com",
        ["user@example.com"],
        connection=connection,
    ).send()

3.4 多 Mailer 架构实战场景

场景一：电商系统——不同邮件用不同通道

# services/email_service.py
from django.core import mail

class EmailService:
    """统一邮件服务，根据业务类型选择不同 mailer"""
    
    MAILER_MAP = {
        'system': 'default',        # 系统通知 → 自建 SMTP
        'marketing': 'marketing',   # 营销邮件 → SendGrid
        'transactional': 'transactional',  # 交易邮件 → 高可靠性 SMTP
    }
    
    @classmethod
    def send(cls, mail_type, subject, body, from_email, to_emails):
        mailer_name = cls.MAILER_MAP.get(mail_type, 'default')
        return mail.send_mail(
            subject, body, from_email, to_emails,
            using=mailer_name,
        )
    
    @classmethod
    def send_order_confirmation(cls, order):
        cls.send(
            'transactional',
            f"订单确认 #{order.id}",
            f"您的订单已确认，预计 {order.estimated_delivery} 到达",
            "order@example.com",
            [order.user.email],
        )
    
    @classmethod
    def send_promotion(cls, user, promotion):
        cls.send(
            'marketing',
            promotion.title,
            promotion.content,
            "marketing@example.com",
            [user.email],
        )

场景二：多区域部署——不同区域用不同邮件服务

# settings/production.py
MAILERS = {
    "default": {
        "BACKEND": "django.core.mail.backends.smtp.EmailBackend",
        "OPTIONS": {"host": "smtp.global.example.com", "use_tls": True},
    },
    "africa": {
        "BACKEND": "example.third_party.EmailBackend",
        "OPTIONS": {"region": "africa-1"},
    },
    "asia": {
        "BACKEND": "example.third_party.EmailBackend",
        "OPTIONS": {"region": "asia-1"},
    },
    "europe": {
        "BACKEND": "django.core.mail.backends.smtp.EmailBackend",
        "OPTIONS": {"host": "smtp.eu.example.com", "use_tls": True},
    },
}

# middleware/select_mailer.py
class RegionMailerMiddleware:
    """根据用户区域自动选择 mailer"""
    
    def __init__(self, get_response):
        self.get_response = get_response
    
    def __call__(self, request):
        # 将当前区域的 mailer 存入 request
        user_region = getattr(request.user, 'region', 'default')
        request.mailer = user_region  # 供后续使用
        return self.get_response(request)

# views.py
from django.core import mail

def send_welcome_email(request, user):
    mail.send_mail(
        "欢迎加入",
        "感谢您注册我们的服务",
        "welcome@example.com",
        [user.email],
        using=request.mailer,  # 使用区域对应的 mailer
    )

场景三：开发/测试环境的 Mailer 切换

# settings/dev.py
MAILERS = {
    "default": {
        "BACKEND": "django.core.mail.backends.filebased.EmailBackend",
        "OPTIONS": {"file_path": "/tmp/dev-mails/"},
    },
    # 开发时也配置其他 mailer，但不真正发送
    "marketing": {
        "BACKEND": "django.core.mail.backends.console.EmailBackend",
    },
}

# settings/test.py
MAILERS = {
    "default": {
        "BACKEND": "django.core.mail.backends.locmem.EmailBackend",
    },
    "marketing": {
        "BACKEND": "django.core.mail.backends.locmem.EmailBackend",
    },
}

3.5 从旧配置迁移到 MAILERS

Django 6.1 提供了平滑过渡机制：MAILERS 和旧的 EMAIL_BACKEND / EMAIL_* 配置可以共存。

# 过渡期：两种配置并存
# 旧的配置仍然有效
EMAIL_BACKEND = 'django.core.mail.backends.smtp.EmailBackend'
EMAIL_HOST = 'smtp.example.com'

# 新的 MAILERS 也可以配置
MAILERS = {
    "marketing": {
        "BACKEND": "sendgrid_backend.SendgridBackend",
        "OPTIONS": {"api_key": "SG.xxx"},
    },
}

# mail.mailers["default"] 会自动使用旧的 EMAIL_BACKEND 配置
# 所以你不需要一次性迁移所有代码

迁移步骤：

在 settings.py 中添加 MAILERS 配置，default 用新的 OPTIONS 格式
旧的 EMAIL_* 配置暂时保留（Django 会发出 deprecation warning）
逐步修改代码中的邮件发送调用，加上 using 参数
完全迁移后，删除旧的 EMAIL_* 配置

# Step 1：添加 MAILERS，保留旧配置
MAILERS = {
    "default": {
        "BACKEND": "django.core.mail.backends.smtp.EmailBackend",
        "OPTIONS": {
            "host": EMAIL_HOST,
            "port": EMAIL_PORT,
            "use_tls": EMAIL_USE_TLS,
            "username": EMAIL_HOST_USER,
            "password": EMAIL_HOST_PASSWORD,
        },
    },
}
# 旧的 EMAIL_* 设置保留，但会有 deprecation warning

# Step 2：逐步修改代码
# 旧代码：
mail.send_mail(subject, body, from_email, recipients)
# 新代码（不改也行，默认用 default mailer）：
mail.send_mail(subject, body, from_email, recipients)

# Step 3：完全迁移后删除旧配置
# 删除 EMAIL_BACKEND、EMAIL_HOST 等所有 EMAIL_* 设置
# 只保留 MAILERS

四、Django 6.1 的其他重要更新

4.1 PBKDF2 迭代次数提升

Django 6.1 将 PBKDF2 密码哈希器的默认迭代次数从 1,200,000 提升到 1,500,000：

# django/contrib/auth/hashers.py
PBKDF2PasswordHasher:
    iterations = 1_500_000  # 从 1_200_000 提升

# 这意味着新注册用户的密码哈希会更安全
# 但也意味着密码验证会慢约 25%（1.5M / 1.2M ≈ 1.25）

生产级注意事项：

# 如果你的登录 API 对延迟敏感，可以考虑：
# 1. 使用 Argon2 哈希器（更安全且可调参数）
PASSWORD_HASHERS = [
    'django.contrib.auth.hashers.Argon2PasswordHasher',  # 首选
    'django.contrib.auth.hashers.PBKDF2PasswordHasher',
    'django.contrib.auth.hashers.PBKDF2SHA1PasswordHasher',
    'django.contrib.auth.hashers.BCryptSHA256PasswordHasher',
]

# 2. 或者自定义 PBKDF2 的迭代次数（不建议降低）
class FastPBKDF2Hasher(PBKDF2PasswordHasher):
    iterations = 1_200_000  # 保持旧值（仅在特殊性能需求时考虑）

4.2 Admin 改进

Django 6.1 的 Admin 有多项改进，其中最值得注意的是表单布局的无障碍优化：

<!-- 旧布局：label 和 input 在同一行 -->
<div class="form-row">
    <label>标题：</label><input type="text" name="title">
</div>

<!-- 新布局：label 在上，input 在下（更符合无障碍规范） -->
<div class="form-row">
    <label>标题：</label>
    <div class="help-text">请输入书籍标题</div>  <!-- help text 在 input 前 -->
    <input type="text" name="title">
</div>

还有几个实用的新特性：

# delete_confirmation_max_display：删除确认页面显示对象数量上限
class BookAdmin(admin.ModelAdmin):
    delete_confirmation_max_display = 50  # 只显示前 50 个对象
    # 默认 None = 不截断（可能显示上千个对象名）

# action 的 location 参数：控制 action 在哪些页面可用
@admin.action(description='标记为已出版', location='change_list')
def mark_published(modeladmin, request, queryset):
    queryset.update(status='published')

@admin.action(description='更新单本书', location='change_form')
def update_single_book(modeladmin, request, obj):
    # 这个 action 只在单对象的编辑页面可用
    obj.status = 'updated'
    obj.save()

# action 的 description_plural 参数
@admin.action(
    description='标记为已出版',
    description_plural='批量标记为已出版',  # 列表页用 plural 描述
)
def mark_published(modeladmin, request, queryset):
    queryset.update(status='published')

4.3 CSP (Content Security Policy) 增强

Django 6.1 对 CSP 支持做了重要增强，新增了 csp_nonce_attr 模板标签：

<!-- 在模板中使用 CSP nonce -->
{% load csp_nonce_attr %}

<script {% csp_nonce_attr %}>
    // 这个 script 标签会自动带上 CSP nonce
    console.log('CSP compliant script');
</script>

<style {% csp_nonce_attr %}>
    body { background: #fff; }
</style>

<!-- 对 Media 对象应用 nonce -->
{{ form.media|csp_nonce_attr }}

CSP nonce 的配置：

# settings.py
INSTALLED_APPS = ['django.middleware.csp']

MIDDLEWARE = [
    'django.middleware.csp.ContentSecurityPolicyMiddleware',
    # ...
]

TEMPLATES = [
    {
        'OPTIONS': {
            'context_processors': [
                'django.template.context_processors.csp',  # 必须配置！
                # ...
            ],
        },
    },
]

# CSP 配置中启用 nonce
CONTENT_SECURITY_POLICY = {
    'directives': {
        'script-src': ['CSP.NONCE', "'self'"],
        'style-src': ['CSP.NONCE', "'self'"],
    },
}

安全警告 W027：如果你启用了 ContentSecurityPolicyMiddleware 并在 CSP 级策略中使用了 CSP.NONCE，但没有配置 django.template.context_processors.csp，Django 会发出 W027 系统检查警告：

$ python manage.py check
?: (security.W027) ContentSecurityPolicyMiddleware is enabled with CSP.NONCE 
   in a CSP policy, but django.template.context_processors.csp is not configured 
   in TEMPLATES. This means CSP nonce attributes will not be available in templates.

4.4 Forms 增强

# Stylesheet 对象：可以给 CSS 链接加自定义属性
from django.forms import Stylesheet

class MyForm(forms.Form):
    class Media:
        css = {
            'all': [
                Stylesheet('css/custom.css', attrs={'crossorigin': 'anonymous'}),
                Stylesheet('css/print.css', media='print'),
            ],
        }

# BLANK_CHOICE_LABEL：更友好的空白选项标签
from django.db.models.fields import BLANK_CHOICE_LABEL

# 旧版：空白选项显示 "---------"
# 新版：空白选项显示更友好、可翻译的文本

# USE_BLANK_CHOICE_DASH：如果需要回退到旧样式
USE_BLANK_CHOICE_DASH = True  # 回退到 "---------"

# FilePathField.set_choices()：动态刷新文件路径选项
class DocumentForm(forms.Form):
    file_path = forms.FilePathField(path='/tmp/documents/')
    
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.fields['file_path'].set_choices()  # 刷新选项列表

4.5 新的模型表达式和函数

# JSONNull：显式表示 JSON null
from django.db.models import JSONNull

class Product(models.Model):
    metadata = models.JSONField()

# 存储 JSON null（区别于 SQL NULL）
Product.objects.create(metadata=JSONNull())  # 存储 JSON null，不是 SQL NULL

# 查询 JSON null
Product.objects.filter(metadata=JSONNull())  # 匹配 JSON null 值
Product.objects.filter(metadata=None)         # 匹配 SQL NULL

# UUID4：自动生成 UUID4 的数据库函数
from django.db.models.functions import UUID4

class Session(models.Model):
    id = models.UUIDField(primary_key=True, default=UUID4, editable=False)

# 或者批量更新
Session.objects.update(id=UUID4())

4.6 Generic Views：RedirectView.preserve_request

from django.views.generic.base import RedirectView

class PreserveMethodRedirectView(RedirectView):
    url = '/new-path/'
    preserve_request = True  # 307/308 代替 302/301
    # POST /old-path → 307 POST /new-path（保留请求方法和 body）
    # 而不是 302 GET /new-path（丢失 body）

五、生产级迁移指南：从 Django 6.0 到 6.1

5.1 升级前检查清单

# 1. 检查 Python 版本
python --version
# Django 6.1 需要 Python 3.12+，推荐 3.14

# 2. 检查依赖兼容性
pip check
# 确保 django-rest-framework、django-filter 等都支持 6.1

# 3. 运行系统检查
python manage.py check --deploy
# 查看所有 security warnings 和 deprecation warnings

# 4. 运行测试
python manage.py test --verbosity=2

# 5. 检查 migration 状态
python manage.py showmigrations

5.2 Fetch Mode 迁移策略

渐进式迁移（推荐）：

# Phase 1：在关键 API 端点使用 FETCH_PEERS
class BookListAPI(APIView):
    def get(self, request):
        # 旧代码：
        # books = Book.objects.all()
        # 新代码：
        books = Book.objects.fetch_mode(models.FETCH_PEERS)
        serializer = BookSerializer(books, many=True)
        return Response(serializer.data)

# Phase 2：在性能关键端点使用 RAISE
class HighFrequencyAPI(APIView):
    def get(self, request):
        books = (
            Book.objects
            .select_related('author', 'category')
            .fetch_mode(models.RAISE)
        )
        # 如果 serializer 访问了未预加载的字段，会报错
        # 这帮助你在开发阶段发现所有 N+1 问题

# Phase 3：全局默认改为 FETCH_PEERS（可选，激进）
# 可以通过自定义 Manager 实现
class FetchPeersManager(models.Manager):
    def get_queryset(self):
        return super().get_queryset().fetch_mode(models.FETCH_PEERS)

class Book(models.Model):
    objects = FetchPeersManager()
    # 所有 Book.objects 查询默认使用 FETCH_PEERS

5.3 on_delete 迁移策略

风险评估矩阵：

场景	迁移到 DB_CASCADE	风险
纯数据级联，无信号依赖	✅ 推荐	低
有 pre_delete/post_delete 信号	❌ 不推荐	信号丢失
关联数据量大 (>10K)	✅ 强烈推荐	Python CASCADE 性能不可接受
需要审计日志	❌ 不推荐	日志丢失
批量清理任务	✅ 推荐	高效