无需GenericFK的Django多态关联（自定义外键）适配遗留PG库

1. 定义类型映射与属性描述符 #

先实现处理模型实例属性读写的描述符，以及模型类与Rails风格类型字符串的双向映射：

from django.db import models
from django.db.models.signals import pre_delete
from django.dispatch import receiver
from django.apps import apps

# 配置模型类与Rails风格类型字符串的映射（替换为你的实际模型路径）
TYPE_MODEL_MAP = {
    "User": "yourapp.models.User",
    "Admin": "yourapp.models.Admin",
    "Client": "yourapp.models.Client",
}

# 构建反向映射：模型类 -> 类型字符串
MODEL_TYPE_MAP = {}
for type_str, model_path in TYPE_MODEL_MAP.items():
    model_cls = apps.get_model(model_path)
    MODEL_TYPE_MAP[model_cls] = type_str

class PolymorphicForeignKeyDescriptor:
    def __init__(self, type_field, id_field):
        self.type_field = type_field
        self.id_field = id_field

    def __get__(self, instance, owner):
        if instance is None:
            return self
        # 从数据库字段读取类型和ID
        type_str = getattr(instance, self.type_field)
        obj_id = getattr(instance, self.id_field)
        if not type_str or not obj_id:
            return None
        
        # 获取对应模型类并查询实例
        model_path = TYPE_MODEL_MAP.get(type_str)
        if not model_path:
            return None
        model_cls = apps.get_model(model_path)
        try:
            return model_cls.objects.get(pk=obj_id)
        except model_cls.DoesNotExist:
            return None

    def __set__(self, instance, value):
        if value is None:
            # 清空关联字段
            setattr(instance, self.type_field, None)
            setattr(instance, self.id_field, None)
            return
        
        # 验证模型类型并设置字段值
        model_cls = type(value)
        type_str = MODEL_TYPE_MAP.get(model_cls)
        if not type_str:
            raise ValueError(f"不支持的模型类型: {model_cls.__name__}")
        
        setattr(instance, self.type_field, type_str)
        setattr(instance, self.id_field, value.pk)

2. 实现自定义字段类 #

编写自定义字段处理查询逻辑，同时整合on_delete的信号处理：

class CustomFK(models.Field):
    def __init__(self, type_field='edited_by_type', id_field='edited_by_id', on_delete=models.DO_NOTHING, **kwargs):
        self.type_field = type_field
        self.id_field = id_field
        self.on_delete = on_delete
        # 该字段不对应实际数据库列，默认允许空值
        kwargs.setdefault('blank', True)
        kwargs.setdefault('null', True)
        super().__init__(**kwargs)

    def get_internal_type(self):
        # 自定义字段标识，不影响核心功能
        return 'CustomFK'

    def contribute_to_class(self, cls, name, private_only=False):
        super().contribute_to_class(cls, name, private_only)
        # 绑定描述符处理属性读写
        setattr(cls, name, PolymorphicForeignKeyDescriptor(self.type_field, self.id_field))
        # 注册信号处理关联对象删除逻辑
        self._register_delete_signals(cls)

    def _register_delete_signals(self, model_cls):
        def handle_related_delete(sender, instance, **kwargs):
            type_str = MODEL_TYPE_MAP.get(sender)
            if not type_str:
                return
            
            filter_kwargs = {
                self.type_field: type_str,
                self.id_field: instance.pk
            }

            # 根据on_delete参数执行对应逻辑
            if self.on_delete == models.CASCADE:
                model_cls.objects.filter(**filter_kwargs).delete()
            elif self.on_delete == models.SET_NULL:
                model_cls.objects.filter(**filter_kwargs).update(**{
                    self.type_field: None,
                    self.id_field: None
                })
            elif self.on_delete == models.DO_NOTHING:
                pass
            # 可按需扩展PROTECT、SET_DEFAULT等逻辑

        # 为所有关联模型注册删除信号
        for model_path in TYPE_MODEL_MAP.values():
            target_model = apps.get_model(model_path)
            pre_delete.connect(handle_related_delete, sender=target_model, weak=False)

    def get_db_prep_lookup(self, lookup_type, value, connection, prepared=False):
        # 处理精确匹配查询
        if lookup_type == 'exact':
            if value is None:
                return [(self.type_field, None), (self.id_field, None)]
            
            type_str = MODEL_TYPE_MAP.get(type(value))
            if not type_str:
                raise ValueError(f"不支持的模型类型: {type(value).__name__}")
            
            return [
                (self.type_field, type_str),
                (self.id_field, value.pk)
            ]
        # 处理IN查询
        elif lookup_type == 'in':
            q_objects = []
            null_included = False
            type_groups = {}

            for obj in value:
                if obj is None:
                    null_included = True
                    continue
                
                model_cls = type(obj)
                type_str = MODEL_TYPE_MAP.get(model_cls)
                if not type_str:
                    raise ValueError(f"不支持的模型类型: {model_cls.__name__}")
                
                if type_str not in type_groups:
                    type_groups[type_str] = []
                type_groups[type_str].append(obj.pk)
            
            # 构建分组OR查询条件
            for type_str, ids in type_groups.items():
                q_objects.append(models.Q(**{
                    self.type_field: type_str,
                    f"{self.id_field}__in": ids
                }))
            # 处理空值情况
            if null_included:
                q_objects.append(models.Q(**{
                    self.type_field: None,
                    self.id_field: None
                }))
            
            return [('OR', q_objects)] if q_objects else []
        
        # 其他查询类型默认不支持，可按需扩展
        raise ValueError(f"不支持的查询类型: {lookup_type}")

    def from_db_value(self, value, expression, connection):
        # 该字段不对应数据库列，由描述符处理实际值读取
        return None

3. 在遗留模型中使用 #

在你的managed=False模型中定义实际数据库字段，再添加自定义多态关联字段：

class MyModel(models.Model):
    # 数据库中已存在的字段
    edited_by_type = models.CharField(max_length=50, blank=True, null=True)
    edited_by_id = models.IntegerField(blank=True, null=True)
    # 自定义多态关联字段
    edited_by = CustomFK(
        type_field='edited_by_type',
        id_field='edited_by_id',
        on_delete=models.SET_NULL
    )

    class Meta:
        managed = False
        db_table = 'your_legacy_table_name'  # 替换为实际表名

4. 使用示例 #

赋值与保存 #

user = User.objects.get(pk=1)
mymodel_obj = MyModel.objects.get(pk=1)
mymodel_obj.edited_by = user
mymodel_obj.save()
# 自动设置edited_by_type="User"，edited_by_id=1

查询操作 #

# 按单个对象过滤
MyModel.objects.filter(edited_by=user)

# 按多个对象过滤
admin = Admin.objects.get(pk=2)
client = Client.objects.get(pk=3)
MyModel.objects.filter(edited_by__in=[user, admin, client])

# 读取关联对象
print(mymodel_obj.edited_by)  # 输出对应的User/Admin/Client实例

注意事项 #

• 确保TYPE_MODEL_MAP中的模型路径与你的项目实际路径一致
• 可根据业务需求扩展on_delete的其他处理逻辑（如PROTECT）
• 如需支持更多查询类型（如__isnull），可扩展get_db_prep_lookup方法

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