在SQLAlchemy中更新父表时同步更新子表字段的实现方法

针对你的需求，有两种常用可行的方案，分别基于数据库原生机制和SQLAlchemy框架的事件系统：

方案一：PostgreSQL触发器（推荐，不受操作方式限制） #

这种方案直接在数据库层面实现联动，不管是通过SQLAlchemy ORM还是直接执行SQL语句修改父表，都会触发子表更新，可靠性更高。

1. 创建触发器函数 #

执行以下SQL语句创建触发器函数，当parent_table的some_val字段更新时，将对应子表的some_val_modified设为True：

CREATE OR REPLACE FUNCTION update_child_modified_flag()
RETURNS TRIGGER AS $$
BEGIN
    -- 仅当父表some_val字段实际变更时，更新子表标记
    IF OLD.some_val IS DISTINCT FROM NEW.some_val THEN
        UPDATE child_table
        SET some_val_modified = TRUE
        WHERE parent_id = NEW.id;
    END IF;
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

2. 绑定触发器到父表 #

将上述函数绑定到parent_table的UPDATE事件上：

CREATE TRIGGER trigger_parent_update_child
AFTER UPDATE OF some_val ON parent_table
FOR EACH ROW
EXECUTE FUNCTION update_child_modified_flag();

你可以在数据库初始化阶段执行这些SQL，也可以通过SQLAlchemy的event.listen钩子，在Base.metadata.create_all完成后自动执行。

方案二：SQLAlchemy事件监听（仅ORM操作生效） #

如果只需要在通过SQLAlchemy ORM操作父表时触发联动，可以使用框架提供的事件监听机制：

修改模型代码并添加事件监听逻辑：

from sqlalchemy import event
from sqlalchemy.orm import Session

class Parent(Base):
    __tablename__ = "parent_table"

    id = mapped_column(Integer, primary_key=True)
    children = relationship("Child", cascade = "all,delete", back_populates="parent")
    some_val = mapped_column(Integer)

class Child(Base):
    __tablename__ = "child_table"

    id = mapped_column(Integer, primary_key=True)
    parent_id = mapped_column(ForeignKey("parent_table.id"))
    parent = relationship("Parent", back_populates="children")
    some_val_modified = mapped_column(Boolean)

# 监听Parent模型的更新事件
@event.listens_for(Parent, "after_update")
def update_child_modified(mapper, connection, target):
    # 检查some_val字段是否发生实际变更
    history = mapper.class_manager["some_val"].history
    if history.has_changes():
        # 批量更新对应子表记录
        session = Session(bind=connection)
        session.query(Child).filter(Child.parent_id == target.id).update(
            {"some_val_modified": True},
            synchronize_session=False
        )
        session.commit()

注意事项 #

• 事件监听方案仅对通过SQLAlchemy ORM执行的更新操作生效，直接用原生SQL修改父表不会触发该逻辑。
• 使用synchronize_session=False可以跳过ORM会话同步，提升批量更新的性能。

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