# apps/catalog/models.py
from apps.catalog.querysets import ProductQuerySet

class Product(TimestampedModel):
    # ... 字段 ...

    # 将 QuerySet 作为默认管理器暴露
    objects = ProductQuerySet.as_manager()

# 干净、可读，不显示 SQL
products = (
    Product.objects
    .active()
    .by_category()"electronics")
    .with_stats()

.in_price_range(min_price=50, max_price=500)
    .select_related("category")
    .order_by("-total_sold")
)

第4步：配置数据库性能设置

# config/settings/production.py
DATABASES = {
    "default": {
        "ENGINE": "django.db.backends.postgresql",
        "NAME": env("DB_NAME"),
        "USER": env("DB_USER"),

{
    "PASSWORD": env("DB_PASSWORD"),
    "HOST": env("DB_HOST"),
    "PORT": env("DB_PORT", default="5432"),
    "CONN_MAX_AGE": 60,           # 连接重用时间为60秒
    "OPTIONS": {

"connect_timeout": 5,
"options": "-c default_transaction_isolation=read committed",
},
}
}

from django.db.models import F

# 不好 — 将对象加载到 Python 中，然后再保存回去
product = Product.objects.get(pk=42)
product.stock -= order_quantity
product.save()  # ← 两次数据库请求。读取和写入之间存在竞争条件。

# 好 — 在数据库层面进行单个原子更新

from django.db.models import Q

# 简单的 OR：处于活动状态或正在促销的产品
products = Product.objects.filter(
    Q(status="active") | Q(on_sale=True)
)

# 复杂的嵌套逻辑
# （活动状态或促销）并且（价格 < 100）并且 NOT（已归档）

products = Product.objects.filter(
    (Q(status="active") | Q(on_sale=True)) &
    Q(price__lt=100) &
    ~Q(status="archived")
)

# 可重用的 Q 对象
def active_and_affordable_q(max_price: float) -> Q:

return Q(status="active") & Q(price__lte=max_price)

budget_products = Product.objects.filter(active_and_affordable_q(50.00))

from django.db.models import (
    Count, Sum, Avg, Min, Max,
    Case, When, Value, IntegerField, DecimalField,
    ExpressionWrapper, F
)

from django.db.models.functions import Coalesce, TruncMonth

# annotate() → 为 QuerySet 中的每个对象添加一个字段
categories_with_counts = Category.objects.annotate(
    product_count=Count("products", distinct=True),
    active_count=Count(
        "products",
        filter=Q(products__status="active"),
        distinct=True,
    ),

avg_price=Coalesce(Avg("products__price"), 0.0),
)
# 每个类别现在都有 .product_count, .active_count, .avg_price

# aggregate() → 返回一个总结整个 QuerySet 的单一字典
summary = Order.objects.filter(
    created_at__year=2025
).aggregate(
    total_revenue=Sum("total"),
    average_order_value=Avg("total"),

order_count=Count(“id“),
max_order=Max(“total“),
min_order=Min(“total“),
)
# 返回: {“total_revenue”: Decimal(“125300.50”), “average_order_value”: …, …}

# 月度收入细分 — 按月分组
monthly_revenue = (
Order.objects
.annotate(month=TruncMonth(“created_at“))

.values("month")
    .annotate(revenue=Sum("total"), orders=Count("id"))
    .order_by("month")
)
# SQL: SELECT DATE_TRUNC('month', created_at) AS month,
#             SUM(total) AS revenue, COUNT(id) AS orders
#      FROM orders_order
#      GROUP BY month ORDER BY month

# 使用 Case/When 进行条件注释
products_with_tier = Product.objects.annotate(
    price_tier=Case(

当(price__lt=25, 则=值(“预算“)),
当(price__lt=100, 则=值(“中档“)),
当(price__lt=500, 则=值(“高端“)),
默认=值(“奢华“),

output_field=models.CharField(),
    )
)
# 每个产品现在都有 .price_tier: "经济型"、"中档"、"高档" 或 "奢华"

# 场景：显示订单及其用户、送货地址和订单项

# 不佳 — N+1 查询

orders = Order.objects.all()
for order in orders:           # 1 查询 (SELECT orders)
    user = order.user          # N 查询 (SELECT user WHERE id=X)
    addr = order.shipping_address  # N 查询
    for item in order.items.all():  # N 查询 (SELECT items WHERE order=X)
        name = item.product.name    # N*M 查询

# 好 — 总共 3 个查询，无论规模如何
orders = (

Order.objects
# select_related: SQL JOIN 用于外键/一对一关系
    .select_related("user", "shipping_address")
# prefetch_related: 针对多对多/反向外键的单独批量查询
    .prefetch_related("items__product__category")
.filter(status="processing")
.order_by("-created_at")
)
# 查询 1: SELECT orders + JOIN users + JOIN addresses WHERE status='processing'
# 查询 2: SELECT items WHERE order_id IN (1, 2, 3, ...)
# 查询 3: SELECT products + JOIN categories WHERE id IN (...)

from django.db.models import Prefetch

# 仅预取活跃的评论，预先排序并过滤

active_reviews = Review.objects.filter(is_approved=True).order_by("-created_at")

products = Product.objects.prefetch_related(
    Prefetch("reviews", queryset=active_reviews, to_attr="approved_reviews")
).active()

# 现在 product.approved_reviews 是一个预加载的列表 — 无需额外查询
for product in products:

top_review = product.approved_reviews[0] if product.approved_reviews else None

from django.db import transaction

# bulk_create: 在一个INSERT语句中插入多行
new_tags = Tag.objects.bulk_create([
    Tag(name="python"),
    Tag(name="django"),
    Tag(name="backend"),
], ignore_conflicts=True)  # ← 如果已存在则跳过
# bulk_update: 在一个UPDATE语句中更新多行

products_to_reactivate = Product.objects.filter(status="archived", stock__gt=0)
for product in products_to_reactivate:
    product.status = "active"

Product.objects.bulk_update(products_to_reactivate, fields=["status"])
# SQL: UPDATE catalog_product SET status=CASE WHEN id=1 THEN 'active' ... END

# update(): 单个 SQL 更新 — 无需模型实例化

Product.objects.filter(
category__slug="electronics",
stock=0,
).update(status="archived")
# SQL: UPDATE catalog_product SET status='archived'
#      WHERE category_id IN (SELECT id FROM category WHERE slug='electronics')
#      AND stock = 0

# 不好 — 一次性加载所有 500,000 行到内存中
for order in Order.objects.filter(status="pending"):
    process_order(order)  # 在规模上存在 OOM 风险

# 好 — 分块获取，内存占用大大降低

for order in Order.objects.filter(status="pending").iterator(chunk_size=2000):
    process_order(order)

# 使用预取（手动，因为 prefetch_related 不适用于迭代器）

for order in Order.objects.filter(status="pending").select_related("user").iterator(chunk_size=2000):
    process_order(order)

在这段代码中，我们使用了一个循环来处理状态为“待处理”的订单。首先，通过 `Order.objects.filter(status=”pending”)` 过滤出所有状态为“待处理”的订单。接着，使用 `select_related(“user”)` 方法来预加载与订单相关的用户信息，以提高查询效率。最后，使用 `iterator(chunk_size=2000)` 方法以每次处理2000个订单的方式进行迭代处理。每个订单都会通过 `process_order(order)` 函数进行处理。

class Order(TimestampedModel):
    # ... 字段 ...

    # ── 模型方法：行级，实例特定逻辑 ──
    def get_subtotal(self) -> Decimal:
        """所有商品价格的总和。需要预先获取商品。"""
        return sum(item.subtotal for item in self.items.all())

def can_be_cancelled(self) -> bool:
        return self.status in ("pending", "processing")

def cancel(self) -> None:
        if not self.can_be_cancelled():

raise ValueError(f"订单 #{self.pk} 无法从状态 '{self.status}'")
self.status = "已取消"
self.save(update_fields=["status", "updated_at"])

# ── 使用管理器/服务进行表级操作 ──
    # 不要在模型方法中放置批量查询或跨表逻辑。

from django.db import transaction

def create_order_with_items(user, cart_items: list) -> Order:
    """
    原子性地创建一个订单及其订单项。
    如果任何操作失败，整个操作将回滚。
    不会有部分订单到达数据库。

with transaction.atomic():
    order = Order.objects.create(
        user=user,
        status=Order.Status.PENDING,
        total=sum(item["price"] * item["quantity"] for item in cart_items),
    )

    order_items = [

OrderItem(
    order=order,
    product_id=item["product_id"],
    quantity=item["quantity"],
    unit_price=item["price"],
)
for item in cart_items
]
OrderItem.objects.bulk_create(order_items)

# 原子性更新库存 — 无竞争条件

        for item in cart_items:
            updated = Product.objects.filter(
                pk=item["product_id"],
                stock__gte=item["quantity"],   # 仅在库存充足时更新
            ).update(stock=F("stock") - item["quantity"])

if updated == 0:
                raise ValueError(f"产品 {item['product_id']}")

        return order

性能优化

1. 使用 only() 和 defer() 限制列的提取

加载不需要的列会浪费 I/O 带宽和内存。 only() 是白名单方法； defer() 是黑名单方法：

# 仅加载您实际使用的列
product_list = (
    Product.objects
    .active()
    .only("id", "name", "slug", "price", "stock")  # ← 5 列而不是 20 列

# 返回字典而不是模型实例——更快，内存占用更少
products = (
    Product.objects
    .active()

# 不好 — 加载整个 QuerySet 来检查存在性
if len(Order.objects.filter(user=user, status="pending")) > 0:
    ...

# 好 — SELECT 1 LIMIT 1 — 没有加载行
if Order.objects.filter(user=user, status="pending").exists():
    ...

# 不好 — 计算 QuerySet

count = len(Product.objects.filter(status="active"))

# 好 — SELECT COUNT(*) — 单个聚合查询
count = Product.objects.filter(status="active").count()

# apps/orders/tests/test_selectors.py
from django.test import TestCase
from apps.orders.selectors import get_orders_for_user
from apps.orders.tests.factories import OrderFactory, OrderItemFactory

class TestOrderSelectorQueryCount(TestCase):
    """
    锁定关键选择器的查询计数。
    如果有人添加了 N+1 查询，这个测试会立即失败
    — 而不是在凌晨 2 点的生产监控中。
    """

def setUp(self):
    self.user = UserFactory()
    orders = OrderFactory.create_batch(10, user=self.user)
    for order in orders:
        OrderItemFactory.create_batch(5, order=order)

def test_get_orders_query_count(self):
    with self.assertNumQueries(3):

这段代码定义了一个测试类的方法，其中 `setUp` 方法用于初始化测试环境，创建一个用户并生成10个订单，每个订单包含5个订单项。接着，`test_get_orders_query_count` 方法用于测试查询的数量，期望在执行过程中只产生3个数据库查询。

# 1: 订单 + JOIN 用户 + JOIN 地址
            # 2: 预取订单项
            # 3: 预取产品
            list(get_orders_for_user(user_id=self.user.id))

class Order(TimestampedModel):
    class Meta:
        indexes = [
            # 复合索引：最常见的查询 — “用户的订单状态”
            models.Index(
                fields=["user", "status"],
                name="idx_order_user_status",
            ),
            # 按创建时间降序排列以便默认排序
            models.Index(
                fields=["-created_at"],

name="idx_order_created_at_desc",
            ),
        ]

        # 部分索引（PostgreSQL）：仅索引未取消的订单
        # 更小、更快的索引，适用于热点查询路径
        # 通过自定义迁移添加 PostgreSQL 特定语法：
        # CREATE INDEX idx_active_orders ON orders_order(user_id)
        # WHERE status != 'cancelled';

# 在 Django shell 中 — 检查慢查询选择器的查询计划
from django.db import connection

with connection.cursor() as cursor:
    cursor.execute("""
        EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
        SELECT * FROM orders_order
        WHERE user_id = 1 AND status = 'processing'
        ORDER BY created_at DESC
        LIMIT 20
    """)
    for row in cursor.fetchall():
        print(row[0])

# 安全 — ORM 对所有值进行参数化

Product.objects.filter(name=user_input)
# SQL: SELECT ... WHERE name = %s -- [user_input]

# 危险 — 原始 SQL 语句与字符串格式化
Product.objects.raw(f"SELECT * FROM catalog_product WHERE name = '{user_input}'")
# ← 永远不要这样做。 user_input = "'; DROP TABLE catalog_product; --"

# 安全 — 原始 SQL 语句与参数化
Product.objects.raw(
    "SELECT * FROM catalog_product WHERE name = %s",
    [user_input]  # ← 始终使用参数化占位符
)

# 危险 — 使用未转义的用户输入调用 extra()
Product.objects.extra(where=[f"name = '{user_input}'"])  # ← 永远不要

# 安全 — 使用参数调用 extra()
Product.objects.extra(where=["name = %s"], params=[user_input])

# 危险 — 攻击者可以设置任何模型字段
order = Order.objects.create(**request.POST.dict())  # ← 切勿这样做

# 安全 — 通过序列化器白名单字段
from apps.orders.serializers import CreateOrderSerializer

serializer = CreateOrderSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
order = OrderService.create(**serializer.validated_data)

3. 通过对象级作用域防止 IDOR

# 危险 — 任何经过身份验证的用户都可以通过 ID 访问任何订单
order = get_object_or_404(Order, pk=pk)  # ← IDOR 漏洞

# 安全 — 始终限制在经过身份验证的用户范围内
order = get_object_or_404(Order, pk=pk, user=request.user)

# 更好 — 在选择器中使用范围查询集

# 不推荐 — 保存所有字段；有覆盖并发更新的风险
product.status = "archived"
product.save()  # ← 重写所有列，包括任何不在内存中的列

# 推荐 — 仅更新你更改的特定列
product.status = "archived"

product.save(update_fields=["status", "updated_at"])
# SQL: UPDATE catalog_product SET status='archived', updated_at=NOW() WHERE id=42

class OrderItem(models.Model):
    quantity   = models.PositiveIntegerField()
    unit_price = models.DecimalField(max_digits=10, decimal_places=2)

    class Meta:
        constraints = [
            models.CheckConstraint(

check=models.Q(quantity__gt=0),
                name="orderitem_positive_quantity",
            ),
            models.CheckConstraint(
                check=models.Q(unit_price__gte=0),
                name="orderitem_non_negative_price",
            ),
            models.UniqueConstraint(

# 不好 — 立即评估 QuerySet，失去链式调用能力
def get_active_products():
    return list(Product.objects.filter(status="active"))  # ← 已评估!

# 无法执行：get_active_products().filter(category=electronics)

# 好 — 返回 QuerySet，让调用者决定何时评估
def get_active_products():

return Product.objects.filter(status="active")  # ← 懒惰，链式调用

# 调用者现在可以进一步筛选：
products = get_active_products().filter(category=electronics).order_by("-price")

# 不好 — 为了计数而将所有行加载到内存中
if len(Order.objects.filter(user=user)) > 0:
    ...
n = len(Product.objects.all())   # 可能加载数百万行

# 好 — 单个 COUNT(*) 查询，加载零行
if Order.objects.filter(user=user).exists():
    ...
n = Product.objects.count()

❌ 错误 3：错误地组合多个注解

使用 annotate() 组合多个聚合会产生错误的结果，因为使用的是连接而不是子查询。

# 不好 — 由于多个 JOIN 导致重复行；author__count 将是错误的
books = Book.objects.annotate(Count("authors"), Count("stores"))
# books[0].authors__count → 错误（膨胀）

# 好 — 使用 distinct=True 来修复重复计数
books = Book.objects.annotate(

author_count=Count("authors", distinct=True),
    store_count=Count("stores", distinct=True),
)

❌ 错误 4：在循环中调用 .save()

# 不好 — 对于 1,000 个产品执行 1,000 次 UPDATE 查询
products = Product.objects.filter(category=electronics)
for product in products:
    product.price *= 0.9  # 10% 折扣
    product.save()        # ← 每个产品一个查询

# 好 — 单个 UPDATE 查询

Product.objects.filter(category=electronics).update(
price=F("price") * Decimal("0.9")
)

# 不好 — 模板循环触发 N+1
{% for order in orders %}
    {{ order.user.email }}      {# 每个订单的数据库访问 #}
    {{ order.items.count() }}   {# 每个订单的数据库访问 #}
{% endfor %}

# 好 — 在传递给模板之前进行预取

orders = Order.objects.select_related("user").prefetch_related("items")

❌ 错误 6：在带注解的 .values() 后使用 .filter()

# 不良示例 — 注解时过滤顺序很重要
# 根据 Django 版本，这可能会产生意想不到的结果
result = (
    Order.objects
    .values("status")
    .annotate(total=Sum("amount"))
    .filter(status="active")  # ← 在这里是安全的，但如果顺序改变则很脆弱
)

# 良好示例 — 尽可能在注解之前进行过滤
result = (
    Order.objects

.filter(status="active")   # ← 首先进行过滤
    .values("status")
    .annotate(total=Sum("amount"))
)

# apps/analytics/selectors.py
from django.db.models import Sum, Count, Avg, F, Q, ExpressionWrapper, DecimalField
from django.db.models.functions import TruncMonth, Coalesce

def get_dashboard_stats(start_date, end_date) -> dict:
    """
    返回完整的仪表板数据，共需4个查询。
    如果采用简单的方法，可能需要100次以上的Python迭代。
    """
    from apps.orders.models import Order, OrderItem
    from apps.catalog.models import Category

    # 查询1：本期按类别的收入
    revenue_by_category = (
        OrderItem.objects
        .filter(order__created_at__range=(start_date, end_date))

.values("product__category__name")
        .annotate(
            revenue=Sum(
                ExpressionWrapper(
                    F("quantity") * F("unit_price"),
                    output_field=DecimalField(max_digits=12, decimal_places=2),
                )
            ),

units_sold=Sum("quantity"),
        )
        .order_by("-revenue")[:10]
    )

    # 查询 2：月度趋势
    monthly_trend = (
        Order.objects
        .filter(created_at__range=(start_date, end_date))
        .annotate(month=TruncMonth("created_at"))

.values("month")
        .annotate(
            revenue=Sum("total"),
            order_count=Count("id"),
            avg_order_value=Avg("total"),
        )
        .order_by("month")
    )

    # 查询 3：按收入排序的热门产品
    top_products = (
        OrderItem.objects

.filter(order__created_at__range=(start_date, end_date))
.values("product__name", "product__id")
.annotate(
    total_revenue=Sum(F("quantity") * F("unit_price"),
                          output_field=DecimalField()),

total_units=Sum("quantity"),
        )
        .order_by("-total_revenue")[:20]
    )

    # 查询 4：低库存警报
    low_stock = (
        Product.objects
        .active()
        .filter(stock__lte=10)

.values("id", "name", "stock", "category__name")
        .order_by("stock")
    )

    return {
        "revenue_by_category": list(revenue_by_category),
        "monthly_trend": list(monthly_trend),
        "top_products": list(top_products),

"low_stock_alerts": list(low_stock),
    }

# apps/exports/services.py

import csv
from django.http import StreamingHttpResponse
from apps.orders.models import Order

class EchoWriter:
    """类似文件的对象，可以回显其输入。用于 StreamingHttpResponse。"""
    def write(self, value):
        return value

def stream_orders_csv(queryset) -> StreamingHttpResponse:
    """
    流式导出大量 CSV 数据，而不将所有内容加载到内存中。
    使用 Django 的 StreamingHttpResponse + ORM 的 iterator() 实现 O(1) 的内存使用。
    """

writer = csv.writer(EchoWriter())

def generate_rows():
    yield writer.writerow(["ID", "用户邮箱", "状态", "总计", "创建时间"])

    for order in (
        queryset
        .select_related("用户")

.only("id", "user__email", "status", "total", "created_at")
            .iterator(chunk_size=2000)
        ):
            yield writer.writerow([
                order.id,
                order.user.email,
                order.status,

str(order.total),
                order.created_at.isoformat(),
            ])

    response = StreamingHttpResponse(generate_rows(), content_type="text/csv")
    response["Content-Disposition"] = 'attachment; filename="orders.csv"'
    return response

# apps/core/managers.py
from django.db import models

class TenantQuerySet(models.QuerySet):
    def for_tenant(self, tenant_id: int):

return self.filter(tenant_id=tenant_id)

class TenantManager(models.Manager):
    def get_queryset(self):
        return TenantQuerySet(self.model, using=self._db)

    def for_tenant(self, tenant_id: int):
        return self.get_queryset().for_tenant(tenant_id)

# apps/projects/models.py
class Project(TimestampedModel):
    tenant = models.ForeignKey("tenants.Tenant", on_delete=models.CASCADE)
    name = models.CharField(max_length=255)
    # ...
    objects = TenantManager()

# 使用 — 在结构上不可能跨越租户边界

projects = Project.objects.for_tenant(request.tenant.id)