Django模型层实现Many2Many关联的公司一致性约束

除了m2m信号，还有以下几种可行方案：

1. 给关联表（through表）添加数据库级约束 #

这是最可靠的兜底方案，直接从数据库层面限制无效数据，不管通过什么方式操作都会被拦截。在M2MTable中添加CheckConstraint，验证员工与部门所属公司一致：

from django.db import models

class M2MTable(models.Model):
    sector = models.ForeignKey(Sector, on_delete=models.CASCADE)
    employee = models.ForeignKey(Employee, on_delete=models.CASCADE)

    class Meta:
        constraints = [
            models.CheckConstraint(
                check=models.Q(sector__company_id=models.F('employee__company_id')),
                name='employee_sector_same_company'
            )
        ]

无论调用Sector.employees.add()还是直接操作M2MTable，只要不符合条件，数据库都会直接抛出错误，从根源阻止无效数据写入。

2. 重写Sector模型中employees关联的add方法 #

自定义多对多关联管理器，覆盖add方法，提前验证要添加的员工是否符合条件：

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

class SectorEmployeeManager(ManyToManyManager):
    def add(self, *objs, **kwargs):
        sector = self.instance
        # 逐个验证员工所属公司与部门是否一致
        for obj in objs:
            if obj.company_id != sector.company_id:
                raise ValueError(f"员工{obj.id}所属公司与部门{sector.id}所属公司不一致")
        # 验证通过后执行原生add逻辑
        super().add(*objs, **kwargs)

class Sector(models.Model):
    company_id = models.ForeignKey(Company, on_delete=models.CASCADE)
    employees = models.ManyToManyField(Employee, through='M2MTable')

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        # 替换关联字段的管理器为自定义实现
        self.employees = SectorEmployeeManager(
            model=Employee,
            instance=self,
            through=M2MTable
        )

这种方式能直接拦截Sector.employees.add()这类操作，提前抛出错误，避免走到数据库层面。

3. 完善through表的save方法并覆盖bulk_create #

如果存在批量创建关联记录的场景，仅重写单个save方法不够，需要覆盖bulk_create实现批量验证：

class M2MTable(models.Model):
    sector = models.ForeignKey(Sector, on_delete=models.CASCADE)
    employee = models.ForeignKey(Employee, on_delete=models.CASCADE)

    def save(self, *args, **kwargs):
        if self.sector.company_id != self.employee.company_id:
            raise ValueError("员工与部门所属公司必须一致")
        super().save(*args, **kwargs)

    @classmethod
    def bulk_create(cls, objs, batch_size=None, ignore_conflicts=False):
        # 批量验证所有待创建的关联记录
        for obj in objs:
            if obj.sector.company_id != obj.employee.company_id:
                raise ValueError(f"员工{obj.employee.id}与部门{obj.sector.id}所属公司不一致")
        super().bulk_create(objs, batch_size, ignore_conflicts)

这样单个保存和批量创建操作都会触发验证逻辑。

推荐优先使用数据库级约束，再配合模型层的方法验证，双重保障既能避免业务代码出错，也能防止绕过模型层的操作（比如直接执行原生SQL）。

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