SQLAlchemy多对多表添加外键约束保证关联关系符合业务规则

我已在Stack Overflow搜索许久，未找到适配我场景的相关方案。

核心实体模型 #

• Persona（包含name、bio字段）
• Episode（包含title、plot字段）
• Clip（包含url、timestamp字段）
• Image（包含url字段）

需要满足的关联约束 #

• 单个Persona可出现在多个Episode中，也可关联这些Episode下的多个Clip、Image，但无需关联对应Episode的所有Clip/Image。
• 单个Episode可包含多个Persona、Clip和Image。
• 单个Image/Clip仅可关联一个Episode，但可关联多个Persona。
• 若Persona已关联若干Episode，该Persona关联的Clip/Image必须属于这些Episode之一，新增Clip/Image也仅能关联该Persona出现过的Episode。
• 若Episode已关联若干Persona，该Episode关联的Clip/Image至少关联其中一个Persona，新增Clip/Image也仅能关联该Episode下的Persona。

现有数据库结构 #

DROP TABLE IF EXISTS episodes;
DROP TABLE IF EXISTS personas;
DROP TABLE IF EXISTS personas_episodes;
DROP TABLE IF EXISTS clips;
DROP TABLE IF EXISTS personas_clips;
DROP TABLE IF EXISTS images;
DROP TABLE IF EXISTS personas_images;


CREATE TABLE episodes (
id INT NOT NULL PRIMARY KEY,
title VARCHAR(120) NOT NULL UNIQUE,
plot TEXT,
tmdb_id VARCHAR(10) NOT NULL,
tvdb_id VARCHAR(10) NOT NULL,
imdb_id VARCHAR(10) NOT NULL);

CREATE TABLE personas (
id INT NOT NULL PRIMARY KEY,
name VARCHAR(30) NOT NULL,
bio TEXT NOT NULL);

CREATE TABLE personas_episodes (
persona_id INT NOT NULL,
episode_id INT NOT NULL,
PRIMARY KEY (persona_id,episode_id),
FOREIGN KEY(persona_id) REFERENCES personas(id),
FOREIGN KEY(episode_id) REFERENCES episodes(id));

CREATE TABLE clips (
id INT NOT NULL PRIMARY KEY,
title VARCHAR(100) NOT NULL,
timestamp VARCHAR(7) NOT NULL,
link VARCHAR(100) NOT NULL,
episode_id INT NOT NULL,
FOREIGN KEY(episode_id) REFERENCES episodes(id));

CREATE TABLE personas_clips (
clip_id INT NOT NULL,
persona_id INT NOT NULL,
PRIMARY KEY (clip_id,persona_id),
FOREIGN KEY(clip_id) REFERENCES clips(id),
FOREIGN KEY(persona_id) REFERENCES personas(id));

CREATE TABLE images (
id INT NOT NULL PRIMARY KEY,
link VARCHAR(120) NOT NULL UNIQUE,
path VARCHAR(120) NOT NULL UNIQUE,
episode_id INT NOT NULL,
FOREIGN KEY(episode_id) REFERENCES episodes(id));

CREATE TABLE personas_images (
persona_id INT NOT NULL,
image_id INT NOT NULL,
PRIMARY KEY (persona_id,image_id),
FOREIGN KEY(persona_id) REFERENCES personas(id),
FOREIGN KEY(image_id) REFERENCES images(id));

现有SQLAlchemy实现代码 #

# db is a configured Flask-SQLAlchemy instance
from app import db
# Alias common SQLAlchemy names
Column = db.Column
relationship = db.relationship


class PkModel(Model):
    """Base model class that adds a 'primary key' column named ``id``."""
 
    __abstract__ = True
    id = Column(db.Integer, primary_key=True)
 

def reference_col(
    tablename, nullable=False, pk_name="id", foreign_key_kwargs=None, column_kwargs=None
):
    """Column that adds primary key foreign key reference.
 
    Usage: ::
 
        category_id = reference_col('category')
        category = relationship('Category', backref='categories')
    """
    foreign_key_kwargs = foreign_key_kwargs or {}
    column_kwargs = column_kwargs or {}
 
    return Column(
        db.ForeignKey(f"{tablename}.{pk_name}", **foreign_key_kwargs),
        nullable=nullable,
        **column_kwargs,
    )

personas_episodes = db.Table(
    "personas_episodes",
    db.Column("persona_id", db.ForeignKey("personas.id"), primary_key=True),
    db.Column("episode_id", db.ForeignKey("episodes.id"), primary_key=True),
)
 
personas_clips = db.Table(
    "personas_clips",
    db.Column("persona_id", db.ForeignKey("personas.id"), primary_key=True),
    db.Column("clip_id", db.ForeignKey("clips.id"), primary_key=True),
)
 
personas_images = db.Table(
    "personas_images",
    db.Column("persona_id", db.ForeignKey("personas.id"), primary_key=True),
    db.Column("image_id", db.ForeignKey("images.id"), primary_key=True),
)
 
 
class Persona(PkModel):
    """One of Roger's personas."""
 
    __tablename__ = "personas"
    name = Column(db.String(80), unique=True, nullable=False)
    bio = Column(db.Text)
    # relationships
    episodes = relationship("Episode", secondary=personas_episodes, back_populates="personas")
    clips = relationship("Clip", secondary=personas_clips, back_populates="personas")
    images = relationship("Image", secondary=personas_images, back_populates="personas")
 
    def __repr__(self):
        """Represent instance as a unique string."""
        return f"<Persona({self.name!r})>"
 
 
class Image(PkModel):
    """An image of one of Roger's personas from an episode of American Dad."""
    
    __tablename__ = "images"
    link = Column(db.String(120), unique=True)
    path = Column(db.String(120), unique=True)
    episode_id = reference_col("episodes")
    # relationships
    personas = relationship("Persona", secondary=personas_images, back_populates="images")
    
 
 
class Episode(PkModel):
    """An episode of American Dad."""
    
    # FIXME: We can add Clips and Images linked to Personas that are not assigned to this episode
 
    __tablename__ = "episodes"
    title = Column(db.String(120), unique=True, nullable=False)
    plot = Column(db.Text)
    tmdb_id = Column(db.String(10))
    tvdb_id = Column(db.String(10))
    imdb_id = Column(db.String(10))
    # relationships
    personas = relationship("Persona", secondary=personas_episodes, back_populates="episodes")
    images = relationship("Image", backref="episode")
    clips = relationship("Clip", backref="episode")
 
    def __repr__(self):
        """Represent instance as a unique string."""
        return f"<Episode({self.title!r})>"
 
 
class Clip(PkModel):
    """A clip from an episode of American Dad that contains one or more of Roger's personas."""
 
    __tablename__ = "clips"
    title = Column(db.String(80), unique=True, nullable=False)
    timestamp = Column(db.String(7), nullable=True)  # 00M:00S
    link = Column(db.String(7), nullable=True)
    episode_id = reference_col("episodes")
    # relationships
    personas = relationship("Persona", secondary=personas_clips, back_populates="clips")

存在的问题 #

当前实现允许向Episode添加关联了不属于该Episode的Persona的Clip和Image，不符合约束规则，找不到合适的方式约束personas+images、personas+clips、personas+episodes三组多对多关联在新增时满足限制条件。
预期行为伪代码如下：

# omitting some detail fields for brevity
e1 = Episode(title="Some Episode")
e2 = Episode(title="Another Episode")
p1 = Persona(name="Raider Dave", episodes=[e1])
p2 = Persona(name="Ricky Spanish", episodes=[e2])
c1 = Clip(title="A clip", episode=e1, personas=[p2])  # should fail
i1 = Image(title="An image", episode=e2, personas=[p1]) # should fail
c2 = Clip(title="Another clip", episode=e1, personas=[p1])  # should succeed
i2 = Image(title="Another image", episode=e2, personas=[p2]) # should succeed

上述示例中c1和i1的创建应该失败，c2和i2的创建应该成功。

可以通过应用层校验+数据库层约束的双重方案实现需求，既保证友好的错误提示，也从根本上避免非法数据写入。

方案1：应用层校验（无需修改表结构，快速实现） #

利用SQLAlchemy的事件监听机制，在数据写入前做规则校验，实现成本最低：

from sqlalchemy import event
from sqlalchemy.exc import ValidationError

# 校验Clip关联规则
@event.listens_for(Clip, 'before_insert')
@event.listens_for(Clip, 'before_update')
def validate_clip_personas(mapper, connection, target):
    if not target.episode_id:
        raise ValidationError("Clip必须关联所属Episode")
    current_episode_id = target.episode_id
    # 校验所有关联的Persona都属于当前Episode
    for persona in target.personas:
        if not any(ep.id == current_episode_id for ep in persona.episodes):
            raise ValidationError(f"Persona {persona.name} 不属于当前Clip所属的Episode，无法关联")
    # 校验至少关联一个Persona
    if len(target.personas) == 0:
        raise ValidationError("Clip至少需要关联一个属于所属Episode的Persona")

# 校验Image关联规则
@event.listens_for(Image, 'before_insert')
@event.listens_for(Image, 'before_update')
def validate_image_personas(mapper, connection, target):
    if not target.episode_id:
        raise ValidationError("Image必须关联所属Episode")
    current_episode_id = target.episode_id
    for persona in target.personas:
        if not any(ep.id == current_episode_id for ep in persona.episodes):
            raise ValidationError(f"Persona {persona.name} 不属于当前Image所属的Episode，无法关联")
    if len(target.personas) == 0:
        raise ValidationError("Image至少需要关联一个属于所属Episode的Persona")

该方案缺点是仅对通过SQLAlchemy写入的数据生效，如果有其他途径直接操作数据库，约束可能失效。

方案2：数据库层约束（PostgreSQL适用，绝对保证规则） #

调整关联表结构，增加复合外键约束，从数据库层面禁止非法关联写入：

1. 先给Clip、Image表增加复合唯一约束，供外键引用：

class Clip(PkModel):
    __tablename__ = "clips"
    title = Column(db.String(80), unique=True, nullable=False)
    timestamp = Column(db.String(7), nullable=True)
    link = Column(db.String(7), nullable=True)
    episode_id = reference_col("episodes")
    # 新增复合唯一约束
    __table_args__ = (db.UniqueConstraint('id', 'episode_id'),)
    personas = relationship("Persona", secondary=personas_clips, back_populates="personas")

class Image(PkModel):
    __tablename__ = "images"
    link = Column(db.String(120), unique=True)
    path = Column(db.String(120), unique=True)
    episode_id = reference_col("episodes")
    # 新增复合唯一约束
    __table_args__ = (db.UniqueConstraint('id', 'episode_id'),)
    personas = relationship("Persona", secondary=personas_images, back_populates="images")

2. 调整多对多关联表结构，增加复合外键：

personas_clips = db.Table(
    "personas_clips",
    db.Column("persona_id", db.ForeignKey("personas.id"), primary_key=True),
    db.Column("clip_id", db.ForeignKey("clips.id"), primary_key=True),
    db.Column("episode_id", db.Integer, nullable=False, primary_key=True),
    # 约束1：保证episode_id和Clip所属的episode一致
    db.ForeignKeyConstraint(
        ['clip_id', 'episode_id'],
        ['clips.id', 'clips.episode_id']
    ),
    # 约束2：保证Persona确实属于该episode
    db.ForeignKeyConstraint(
        ['persona_id', 'episode_id'],
        ['personas_episodes.persona_id', 'personas_episodes.episode_id']
    )
)

personas_images = db.Table(
    "personas_images",
    db.Column("persona_id", db.ForeignKey("personas.id"), primary_key=True),
    db.Column("image_id", db.ForeignKey("images.id"), primary_key=True),
    db.Column("episode_id", db.Integer, nullable=False, primary_key=True),
    db.ForeignKeyConstraint(
        ['image_id', 'episode_id'],
        ['images.id', 'images.episode_id']
    ),
    db.ForeignKeyConstraint(
        ['persona_id', 'episode_id'],
        ['personas_episodes.persona_id', 'personas_episodes.episode_id']
    )
)

后续新增关联时只需要同步填充episode_id字段即可，数据库会自动拦截所有不符合规则的写入操作。

生产环境建议两种方案结合使用，应用层做前置校验返回友好提示，数据库层做最终兜底。

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