自关联外键场景下如何复用现有查询获取父主体及子级所有关联资产

解决方案 #

有两种改动极小的方案可以实现你的需求，完全不需要修改现有业务查询代码，无需大规模重构： #

方案一：重写Asset默认管理器（最易用，兼容性最好） #

只需要修改两个模型的定义，原有Asset.objects.filter(owner=parent)的写法完全生效：

1. 首先完善Owner模型，新增获取自身+所有子主体的方法：

from django.db import models

class Owner(models.Model):
    parent = models.ForeignKey(
        "self",
        on_delete=models.CASCADE, # Django 2.x+ 外键必填参数
        null=True, blank=True, # 根节点无父级，允许为空
        related_name="children"
    )

    def get_descendants_including_self(self):
        # 层级少（3级以内）可直接用Python递归
        res = [self]
        for child in self.children.all():
            res.extend(child.get_descendants_including_self())
        return res
    
    # 层级深/数据量大的场景，推荐用数据库递归CTE（PostgreSQL为例），性能更好
    # def get_descendants_including_self(self):
    #     from django.db import connection
    #     with connection.cursor() as cursor:
    #         cursor.execute("""
    #             WITH RECURSIVE owner_tree AS (
    #                 SELECT id FROM your_app_name_owner WHERE id = %s
    #                 UNION ALL
    #                 SELECT o.id FROM your_app_name_owner o
    #                 JOIN owner_tree ot ON o.parent_id = ot.id
    #             )
    #             SELECT id FROM owner_tree
    #         """, [self.id])
    #         return [row[0] for row in cursor.fetchall()]

2. 自定义Asset的查询集和默认管理器，自动处理owner查询参数：

from django.db.models import Manager, QuerySet

class AssetQuerySet(QuerySet):
    def filter(self, *args, **kwargs):
        # 拦截owner等于查询，替换为包含所有子主体的IN查询
        if "owner" in kwargs and isinstance(kwargs["owner"], Owner):
            target_owner = kwargs.pop("owner")
            kwargs["owner__in"] = target_owner.get_descendants_including_self()
        return super().filter(*args, **kwargs)

class AssetManager(Manager):
    def get_queryset(self):
        return AssetQuerySet(self.model, using=self._db)

class Asset(models.Model):
    owner = models.ForeignKey(
        Owner,
        on_delete=models.CASCADE
    )
    # 替换默认管理器，所有现有调用Asset.objects的逻辑自动生效
    objects = AssetManager()

方案二：注册自定义Lookup（更通用，适配所有关联Owner的模型） #

如果你还有其他模型也关联了Owner外键，需要统一实现相同的查询逻辑，可以直接覆盖Owner外键的默认等于匹配规则：

from django.db.models import Lookup
from django.db.models.fields.related import ForeignKey

# 自定义等于查找逻辑
class OwnerEqualLookup(Lookup):
    lookup_name = "exact"

    def as_sql(self, compiler, connection):
        lhs, lhs_params = self.process_lhs(compiler, connection)
        rhs, rhs_params = self.process_rhs(compiler, connection)
        params = lhs_params + rhs_params
        # 直接使用数据库递归CTE实现查询
        return f"""
        {lhs} IN (
            WITH RECURSIVE owner_tree AS (
                SELECT id FROM your_app_name_owner WHERE id = %s
                UNION ALL
                SELECT o.id FROM your_app_name_owner o
                JOIN owner_tree ot ON o.parent_id = ot.id
            )
            SELECT id FROM owner_tree
        )
        """, params

# 注册Lookup到Owner类型的外键，全局生效
ForeignKey.register_lookup(OwnerEqualLookup, for_model=Owner)

注意事项 #

• 上述两种方案均兼容所有现有查询写法，无需修改业务代码
• 如果使用MySQL 8.0以下版本不支持CTE，建议使用Python递归方案，或提前缓存所有子主体ID优化性能

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