SQLAlchemy多关联表字段约束：限制D表a_id关联合法性

要实现table_D.a_id的特殊约束，确保其关联的A记录与当前D记录关联的C记录属于同一个B，可以通过数据库层面的CHECK约束（推荐，强制且无法绕过）或ORM层面的验证（作为补充）来实现。

一、数据库层面：添加CHECK约束 #

这是最可靠的方式，直接在数据库表中定义约束，任何违反约束的操作都会被数据库拒绝。

修改代码步骤： #

1. 导入CheckConstraint类；
2. 在D类中添加__table_args__属性，定义CHECK约束逻辑。

修改后的D类代码：

from sqlalchemy import CheckConstraint

class D(Base):
    __tablename__ = 'd'
    id: Mapped[int] = mapped_column(primary_key=True)

    c_id: Mapped[int] = mapped_column(ForeignKey('c.id'))
    c: Mapped['C'] = relationship(back_populates='ds')

    a_id: Mapped[Optional[int]] = mapped_column(ForeignKey('a.id'))
    a: Mapped['A'] = relationship(secondary='join(B,C,B.id==C.b_id)',
                                  secondaryjoin='A.b_id==B.id',
                                  primaryjoin='C.id==D.c_id',
                                    viewonly=True)
    
    __table_args__ = (
        CheckConstraint(
            """
            a_id IS NULL 
            OR EXISTS (
                SELECT 1
                FROM a
                JOIN b ON a.b_id = b.id
                JOIN c ON c.b_id = b.id
                WHERE a.id = d.a_id AND c.id = d.c_id
            )
            """,
            name='d_a_id_valid'
        ),
    )

注意事项： #

• 若使用SQLite，需确保版本≥3.37.0（2021-11-27发布），旧版本SQLite会忽略CHECK约束；
• 若使用PostgreSQL、MySQL（8.0.16+）等数据库，CHECK约束均被原生支持。

当执行你的测试代码时，插入d2会触发数据库约束错误，阻止提交，符合预期。

二、ORM层面：添加验证逻辑 #

作为数据库约束的补充，可以在ORM层面对实例进行验证，提前拦截非法数据。使用validates装饰器实现：

from sqlalchemy.orm import validates

class D(Base):
    __tablename__ = 'd'
    id: Mapped[int] = mapped_column(primary_key=True)

    c_id: Mapped[int] = mapped_column(ForeignKey('c.id'))
    c: Mapped['C'] = relationship(back_populates='ds')

    a_id: Mapped[Optional[int]] = mapped_column(ForeignKey('a.id'))
    a: Mapped['A'] = relationship(secondary='join(B,C,B.id==C.b_id)',
                                  secondaryjoin='A.b_id==B.id',
                                  primaryjoin='C.id==D.c_id',
                                    viewonly=True)
    
    @validates('a', 'c')
    def validate_a_c_b_match(self, key, value):
        # 当a或c被设置时，验证两者的b_id是否一致
        if key == 'a' and self.c is not None:
            if value.b_id != self.c.b_id:
                raise ValueError("A和C必须属于同一个B")
        elif key == 'c' and self.a is not None:
            if value.b_id != self.a.b_id:
                raise ValueError("A和C必须属于同一个B")
        return value

这种方式在创建或修改D实例时会立即验证，但无法拦截直接通过SQL执行的操作，因此建议和数据库约束配合使用。

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