如何在MySQL、PostgreSQL及Django中保证表间关联数据一致性？

1. 核心校验规则实现 #

要确保表A中(prof_id, subj_id)组合（当prof_id非空时）必须存在于ProfessorsSubject表中，可按以下思路处理：

1.2 仅通过字段属性和约束实现（优先方案） #

多数现代关系型数据库（PostgreSQL、MySQL 8.0+等）支持直接通过约束实现需求，无需额外逻辑：

• 复合外键约束：若ProfessorsSubject表的主键（或唯一键）为(prof_id, subj_id)，可给表A添加复合外键指向该组合。由于SQL中外键约束对NULL值自动跳过校验，刚好适配prof_id可为空的业务规则。
  示例SQL（PostgreSQL）：

  ALTER TABLE table_a
  ADD CONSTRAINT fk_table_a_prof_subj
  FOREIGN KEY (prof_id, subj_id)
  REFERENCES professors_subject(prof_id, subj_id);
  

• 条件CHECK约束（适用于不支持复合外键或需更灵活规则的场景）：

  ALTER TABLE table_a
  ADD CONSTRAINT chk_table_a_prof_subj_valid
  CHECK (
    prof_id IS NULL
    OR EXISTS (
      SELECT 1 FROM professors_subject
      WHERE prof_id = table_a.prof_id AND subj_id = table_a.subj_id
    )
  );
  

  注意：MySQL 5.x版本的CHECK约束不生效，需用触发器替代。

同时，表A要求的(sclass_id, subj_id)唯一约束可直接添加：

ALTER TABLE table_a
ADD CONSTRAINT uq_table_a_sclass_subj UNIQUE (sclass_id, subj_id);

1.1 是否需要存储过程或触发器？ #

仅当数据库不支持上述约束（如旧版MySQL）时，才需要用触发器实现校验：

• 示例触发器（MySQL）：

  DELIMITER //
  CREATE TRIGGER trg_table_a_before_insert
  BEFORE INSERT ON table_a
  FOR EACH ROW
  BEGIN
    IF NEW.prof_id IS NOT NULL THEN
      IF NOT EXISTS (
        SELECT 1 FROM professors_subject
        WHERE prof_id = NEW.prof_id AND subj_id = NEW.subj_id
      ) THEN
        SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = '该教授无权教授此科目';
      END IF;
    END IF;
  END //
  DELIMITER ;
  

  存储过程用于批量或复杂业务逻辑，此处无需优先考虑，触发器即可满足插入/更新前的校验需求。

Django原生ORM对复合外键和条件约束的支持有限，需通过以下方式实现数据库层面的强制限制：

1. 自定义约束实现（适用于支持CHECK的数据库） #

通过模型的Meta.constraints添加原生SQL约束，确保规则生效在数据库层面：

from django.db import models

class TableA(models.Model):
    sclass = models.ForeignKey(StudentClass, on_delete=models.CASCADE, db_column='sclass_id')
    subj = models.ForeignKey(Subject, on_delete=models.CASCADE, db_column='subj_id')
    prof = models.ForeignKey(Professor, on_delete=models.SET_NULL, null=True, blank=True, db_column='prof_id')

    class Meta:
        constraints = [
            # 实现(sclass_id, subj_id)唯一约束
            models.UniqueConstraint(fields=['sclass', 'subj'], name='uq_table_a_sclass_subj'),
            # 实现(prof_id, subj_id)的合法性校验
            models.CheckConstraint(
                check=models.RawSQL(
                    """
                    prof_id IS NULL
                    OR EXISTS (
                        SELECT 1 FROM professors_subject
                        WHERE prof_id = table_a.prof_id AND subj_id = table_a.subj_id
                    )
                    """,
                    []
                ),
                name='chk_table_a_prof_subj_valid'
            )
        ]

2. 触发器实现（适用于旧版MySQL） #

通过Django迁移文件的RunSQL操作创建触发器：

from django.db import migrations

class Migration(migrations.Migration):
    dependencies = [
        ('your_app', '0001_initial'),
    ]

    operations = [
        # 插入前校验触发器
        migrations.RunSQL("""
            DELIMITER //
            CREATE TRIGGER trg_table_a_before_insert
            BEFORE INSERT ON your_app_tablea
            FOR EACH ROW
            BEGIN
                IF NEW.prof_id IS NOT NULL THEN
                    IF NOT EXISTS (
                        SELECT 1 FROM your_app_professorsubject
                        WHERE prof_id = NEW.prof_id AND subj_id = NEW.subj_id
                    ) THEN
                        SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = '该教授无权教授此科目';
                    END IF;
                END IF;
            END //
            DELIMITER ;
        """),
        # 更新前校验触发器
        migrations.RunSQL("""
            DELIMITER //
            CREATE TRIGGER trg_table_a_before_update
            BEFORE UPDATE ON your_app_tablea
            FOR EACH ROW
            BEGIN
                IF NEW.prof_id IS NOT NULL THEN
                    IF NOT EXISTS (
                        SELECT 1 FROM your_app_professorsubject
                        WHERE prof_id = NEW.prof_id AND subj_id = NEW.subj_id
                    ) THEN
                        SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = '该教授无权教授此科目';
                    END IF;
                END IF;
            END //
            DELIMITER ;
        """)
    ]

内容的提问来源于stack exchange，提问作者Vinícius Costa