Django：如何通过ORM或MySQL过滤自关联模型自定义property？

现有实现的问题根源 #

原始模型定义如下：

class Production(models.Model):
    copy = models.ForeignKey(
        'self',
        related_name='children',
        on_delete=models.SET_NULL,
        blank=True,
        null=True
    )

    @property
    def num(self):
        n = 0
        if self.copy:
            n += 1
            n += self.copy.num
        return n

对应的低效查询代码：

# 现有实现
qs = [q for q in Production.objects.filter(copy__isnull=False) if q.num > 2]

这套实现的性能损耗核心是触发了N+1查询问题：先批量拉取所有存在复制关联的对象，再逐个访问Python层的num属性，每次访问属性都会沿着外键链条递归查询数据库，1000条数据最多可能触发上万次数据库交互，数据量越大性能衰减越明显。

需求本质是筛选复制链路长度超过2、即从源对象开始经过3次及以上复制生成的Production记录。

可落地的优化路径 #

方案1：基于递归CTE的单查询实现（MySQL 8.0+适用） #

MySQL 8.0及以上版本支持递归公共表达式（CTE），可以把原本在Python层的递归计算完全下推到数据库执行，仅需一次查询就能完成全链路深度计算，从根源上消除N+1问题。
通过Django原生查询接口实现的代码如下：

from django.db import connection

def fetch_deep_copy_productions():
    recursive_cte = """
        production_copy_depth AS (
            -- 锚点：原始无复制来源的对象，链路深度为0
            SELECT id, copy_id, 0 AS depth
            FROM production_production
            WHERE copy_id IS NULL

            UNION ALL

            -- 递归：沿copy外键向上累加链路深度
            SELECT p.id, p.copy_id, cte.depth + 1 AS depth
            FROM production_production p
            INNER JOIN production_copy_depth cte ON p.copy_id = cte.id
        )
        SELECT p.* FROM production_production p
        INNER JOIN production_copy_depth cte ON p.id = cte.id
        WHERE cte.depth > 2
    """
    with connection.cursor() as cursor:
        cursor.execute(f"WITH RECURSIVE {recursive_cte}")
        columns = [col[0] for col in cursor.description]
        result = [
            dict(zip(columns, row))
            for row in cursor.fetchall()
        ]
    return result

• 该方案不需要修改原有表结构，所有递归计算都在数据库内存中完成，查询效率随数据量增长的衰减非常平缓。

方案2：冗余字段优化（适配低版本MySQL，兼顾查询性能） #

如果使用的MySQL版本低于8.0不支持递归CTE，可以在模型上新增带索引的冗余字段存储复制链路深度，把计算成本转移到写入阶段：

1. 修改模型定义：

   class Production(models.Model):
       copy = models.ForeignKey('self', related_name='children', on_delete=models.SET_NULL, blank=True, null=True)
       # 新增冗余字段，记录当前对象的复制链路深度，加索引提升查询效率
       copy_depth = models.IntegerField(default=0, db_index=True)
   

2. 重写模型的save方法，在写入时自动同步深度值，和原num属性的计算逻辑保持完全一致：

   def save(self, *args, **kwargs):
       if self.copy:
           self.copy_depth = self.copy.copy_depth + 1
       else:
           self.copy_depth = 0
       super().save(*args, **kwargs)
   

3. 后续查询直接走数据库索引过滤，单条SQL就能拿到结果：

   qs = Production.objects.filter(copy_depth__gt=2)
   

• 该方案的查询性能是所有方案中最高的，仅需维护写入阶段的深度同步逻辑，适合读写比高、查询频繁的场景。

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