如何重写Django ManyRelatedManager方法并实现关联验证

问题描述 #

现有两个Django模型Category和Field，二者为多对多关联，需要实现以下约束：

• 调用管理器add()方法建立关联时，若Field为受控字段（即controller不为空），其控制器字段必须已关联到该Category；
• 调用set()方法时，列表中所有受控Field的控制器字段需在列表内，或已关联到该Category。

模型代码：

class Category(models.Model):
    name = models.CharField(max_length=50, unique=True)
    slug = models.SlugField(max_length=50, unique=True, editable=False)
    fields = models.ManyToManyField("Field", related_name="categories")

class Field(models.Model):
    controller = models.ForeignKey(
        "self",
        models.CASCADE,
        related_name="dependents",
        null=True,
        blank=True,
    )
    label = models.CharField(max_length=50)
    name = models.CharField(max_length=50)

示例场景：

# 创建Category
category = Category.objects.create(name="Cars", slug="cars")
# 创建Field
brand_field = Field.objects.create(name="brand", label="Brand")
model_field = Field.objects.create(name="model", label="Model", controller=brand_field)
# 调用add时，控制器需已关联，否则抛异常
category.fields.add(model_field)
# 调用set时，控制器需在列表内或已关联，否则抛异常
category.fields.set([model_field])

解决方案一：重写ManyRelatedManager #

直接修改多对多字段的关联管理器，在add()和set()方法中嵌入校验逻辑，确保任何代码调用都会触发约束检查。

1. 自定义关联管理器 #

继承默认的ManyRelatedManager，重写核心方法：

from django.db.models.fields.related import ManyRelatedManager

class ConstrainedManyRelatedManager(ManyRelatedManager):
    def add(self, *objs, through_defaults=None):
        # 校验每个待添加的受控Field
        for obj in objs:
            if obj.controller is not None:
                if obj.controller not in self.instance.fields.all():
                    raise ValueError(f"字段 {obj.name} 的控制器 {obj.controller.name} 未关联当前分类")
        # 调用父类方法完成添加
        super().add(*objs, through_defaults=through_defaults)

    def set(self, objs, clear=False, through_defaults=None):
        # 整理所有相关字段：已关联的 + 待设置的（如果不清除原有关联）
        existing_fields = set(self.instance.fields.all())
        new_fields = set(objs)
        all_valid_fields = existing_fields.union(new_fields) if not clear else new_fields

        # 校验每个待设置的受控Field
        for obj in new_fields:
            if obj.controller is not None:
                if obj.controller not in all_valid_fields:
                    raise ValueError(f"字段 {obj.name} 的控制器 {obj.controller.name} 不在待设置列表中，也未关联当前分类")
        # 调用父类方法完成设置
        super().set(objs, clear=clear, through_defaults=through_defaults)

2. 自定义多对多字段 #

创建自定义ManyToManyField，指定使用上面的管理器：

from django.db.models import ManyToManyField

class ConstrainedManyToManyField(ManyToManyField):
    def _get_manager(self):
        return ConstrainedManyRelatedManager(
            model=self.related_model,
            instance=self.instance,
            through=self.remote_field.through,
            reverse=False,
        )

3. 修改Category模型 #

替换原有的多对多字段为自定义字段：

class Category(models.Model):
    name = models.CharField(max_length=50, unique=True)
    slug = models.SlugField(max_length=50, unique=True, editable=False)
    fields = ConstrainedManyToManyField("Field", related_name="categories")

这样无论在代码中直接调用add()/set()，还是通过其他方式操作关联，都会自动执行校验，不符合条件时抛出ValueError。

解决方案二：在ModelAdmin中拦截保存操作 #

如果只需要在后台管理界面管控关联逻辑，可以在ModelAdmin中拦截保存动作，校验通过后再执行保存。

实现代码 #

from django.contrib import admin
from .models import Category, Field

class CategoryAdmin(admin.ModelAdmin):
    def save_model(self, request, obj, form, change):
        # 处理修改场景：对比原有关联和新关联字段
        if change:
            original_fields = set(Category.objects.get(pk=obj.pk).fields.all())
            new_fields = set(form.cleaned_data['fields'])
            # 校验新增的受控字段
            for field in new_fields - original_fields:
                if field.controller is not None:
                    if field.controller not in new_fields and field.controller not in original_fields:
                        self.message_user(request, f"字段 {field.name} 的控制器 {field.controller.name} 未关联或未选中", level='error')
                        return
        # 处理新增场景：直接校验选中的字段
        else:
            for field in form.cleaned_data['fields']:
                if field.controller is not None:
                    if field.controller not in form.cleaned_data['fields']:
                        self.message_user(request, f"字段 {field.name} 的控制器 {field.controller.name} 未被选中", level='error')
                        return
        # 校验通过，执行保存
        super().save_model(request, obj, form, change)

admin.site.register(Category, CategoryAdmin)
admin.site.register(Field)

注意：这种方式仅对后台管理界面的操作生效，如果代码中直接调用add()/set()，不会触发校验。

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