Django单关联模型加过滤时prefetch_related仍触发N+1查询问题

之前的prefetch_related不生效，核心是两个写法问题：

• 预取层级错误：最开始的Profile.objects.all()没有预取关联的signals，遍历每个profile取profile.signals时就已经产生了第一轮N次查询。把prefetch_related("orders")写在循环内的profile.signals查询上，等于每个profile查询signals时都要单独做一次orders预取，查询量直接翻倍。
• 预取缓存被绕过：Django的prefetch_related缓存仅在直接访问关联集合、调用all()时生效，一旦对关联字段链式调用.filter()、.order_by()等方法生成新查询集，ORM不会复用预取缓存，会直接发起新的数据库查询。示例代码中对signal.orders的四次过滤、排序调用，每一次都会触发独立SQL，预取逻辑完全失效。

根据数据量级选以下两种方案即可彻底解决N+1问题： #

方案1：自定义Prefetch对象（推荐，数据量大时优先选） #

从最外层查询开始把两层关联的预取逻辑全部定义好，通过to_attr参数把不同过滤规则的结果直接绑定到Signal实例的自定义属性上，循环内直接读取属性，全程不会触发额外查询。

from django.db.models import Prefetch

# 预取阶段就定义好所有需要的Order过滤规则
enter_orders_sorted = Order.objects.filter(
    position_direction=OrderDirectionChoice.ENTER
).order_by("exchanged_at")
enter_orders = Order.objects.filter(position_direction=OrderDirectionChoice.ENTER)
exit_orders = Order.objects.filter(position_direction=OrderDirectionChoice.EXIT)
exchanged_orders = Order.objects.filter(exchanged_at__isnull=False)

# 最外层查询一次性完成所有关联预取
profiles = Profile.objects.prefetch_related(
    Prefetch(
        "signals",
        queryset=Signal.objects.prefetch_related(
            Prefetch("orders", queryset=enter_orders_sorted, to_attr="filter1_res"),
            Prefetch("orders", queryset=enter_orders, to_attr="filter2_res"),
            Prefetch("orders", queryset=exit_orders, to_attr="filter3_res"),
            Prefetch("orders", queryset=exchanged_orders, to_attr="filter4_res"),
        )
    )
).all()

for profile in profiles:
    for signal in profile.signals.all():
        # 直接读取预取阶段计算好的属性，无额外SQL
        print(signal.filter1_res, signal.filter2_res, signal.filter3_res, signal.filter4_res)

方案2：全量预取+内存过滤（适合单Signal关联Order量少的场景） #

如果每个Signal下关联的Order数量不多，可以直接用双下划线预取两层关联，拿到全量Order数据后在Python内存里做过滤排序，总SQL次数更少，代码更简洁。

# 总共3条SQL完成全量数据查询：1查Profile，1查关联Signal，1查关联Order
profiles = Profile.objects.prefetch_related("signals__orders").all()

for profile in profiles:
    for signal in profile.signals.all():
        # 调用all()命中预取缓存，不会触发数据库查询
        all_orders = list(signal.orders.all())
        filter1 = sorted(
            [o for o in all_orders if o.position_direction == OrderDirectionChoice.ENTER],
            key=lambda x: x.exchanged_at
        )
        filter2 = [o for o in all_orders if o.position_direction == OrderDirectionChoice.ENTER]
        filter3 = [o for o in all_orders if o.position_direction == OrderDirectionChoice.EXIT]
        filter4 = [o for o in all_orders if o.exchanged_at is not None]
        print(filter1, filter2, filter3, filter4)

方案选择参考 #

• 单Signal关联Order数量超过50、或者Order表总数据量超过10w，选方案1，在数据库层面做过滤，减少无效数据传输和内存占用
• 数据量小的场景选方案2，总SQL数更少，代码维护成本低

内容的提问来源于stack exchange，提问作者unknown