如何为Django QuerySet预取JSONField中ID关联的Environment对象？

背景 #

我有一个复杂分析应用，原本采用传统关系型架构，用CRUD模型更新实体。但系统高度事件驱动，历史记录管理和数据迁移难度极大，因此转向类NoSQL方案：用JSON存储数据，通过补丁链表记录历史。这个方案在数据结构上很优雅，也能适配其他服务，但在Django ORM（搭配PostgreSQL）中遇到了关联查询的问题。

现状 #

原模型伪代码：

class Environment(models.Model):
    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)

class Analysis(models.Model):
    created = models.DateTime(auto_now_add=True)
    environment = models.ForeignKey(
        "myapp.Environment",
        blank=True,
        null=True,
        related_name="analyses",
    )
    everything_else = models.JSONField()

调整后的模型：

class Environment(models.Model):  
    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)  
  
class Analysis(models.Model):  
    created = models.DateTime(auto_now_add=True)
    everything = models.JSONField()

everything字段的JSON结构示例：

{
  "environment": "2bf94e55-7b47-41ad-b81a-6cce59762160",
  "other_stuff": "lots of stuff"
}

现在通过ID仍能获取Environment对象，但执行以下代码会触发N+1查询：

for analysis in Analysis.objects.filter(created=today).all():
   print(Environment.objects.get(id=analysis.everything['environment']))

手动收集ID再批量查询的方法无法适配依赖get_queryset的Django组件（如API序列化器、Admin），需要能直接整合进QuerySet的解决方案。

解决方案 #

方案1：Python层面批量预取（推荐，适配Django生态） #

通过自定义QuerySet方法，在Python层面完成批量查询和关联绑定，无需修改数据库结构，完美兼容序列化器和Admin：

1. 自定义Analysis的QuerySet：

import uuid
from django.db.models import QuerySet

class AnalysisQuerySet(QuerySet):
    def prefetch_environments(self):
        # 先获取当前QuerySet的所有分析对象
        analysis_list = list(self)
        # 提取所有非空的环境ID
        env_ids = []
        for analysis in analysis_list:
            env_id_str = analysis.everything.get("environment")
            if env_id_str:
                try:
                    env_ids.append(uuid.UUID(env_id_str))
                except ValueError:
                    continue  # 处理无效UUID的情况
        # 批量查询环境，用in_bulk提高效率
        env_map = Environment.objects.in_bulk(env_ids)
        # 给每个分析对象绑定对应的环境实例
        for analysis in analysis_list:
            env_id_str = analysis.everything.get("environment")
            if env_id_str:
                try:
                    analysis.environment = env_map.get(uuid.UUID(env_id_str))
                except ValueError:
                    analysis.environment = None
            else:
                analysis.environment = None
        return analysis_list

2. 把自定义QuerySet绑定到Analysis模型：

class Analysis(models.Model):  
    created = models.DateTime(auto_now_add=True)
    everything = models.JSONField()
    
    objects = AnalysisQuerySet.as_manager()

3. 使用方式：

# 直接在QuerySet后调用prefetch_environments即可
analyses = Analysis.objects.filter(created=today).prefetch_environments()

# 遍历或序列化时直接访问analysis.environment，不会触发额外查询
for analysis in analyses:
    print(analysis.environment.id if analysis.environment else "无环境")

这个方法的优势是：

• 完全兼容Django的序列化器（只需在序列化类中添加environment字段即可）
• 适配Admin界面（可在list_display中直接使用environment）
• 逻辑简单，无需依赖PostgreSQL的特定语法

方案2：数据库层面注解（适合复杂过滤场景） #

利用PostgreSQL对JSONB的支持，通过ORM注解提取环境ID，再用子查询关联环境数据，适合需要在数据库层面过滤关联环境的场景：

import uuid
from django.db import models
from django.db.models import Subquery, OuterRef
from django.contrib.postgres.fields.jsonb import KeyTextTransform
from django.contrib.postgres.fields import UUIDField

# 提取JSON中的环境ID并转为UUID类型
analyses = Analysis.objects.filter(created=today).annotate(
    environment_id=models.Cast(
        KeyTextTransform("environment", "everything"),
        output_field=UUIDField()
    )
)

# 用子查询预取环境的特定字段（如需完整对象，需逐个字段注解）
env_subquery = Environment.objects.filter(
    id=OuterRef("environment_id")
).values("id", "name")[:1]  # 替换为你需要的字段

analyses = analyses.annotate(
    environment_id=models.F("environment_id"),
    environment_name=Subquery(env_subquery.values("name"), output_field=models.CharField()),
    environment_uuid=Subquery(env_subquery.values("id"), output_field=UUIDField())
)

这种方法的优势是能在数据库层面完成关联过滤，但缺点是无法直接获取完整的Environment对象，需逐个字段注解，适合只需要环境部分字段的场景。

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