SQLAlchemy插入前修改同表对象未持久化问题求助

尝试为scenario_training_schedule表添加事件监听器，插入新行时将同scenario_id下此前激活的行设置为非激活状态（active=False）。调试时通过断点可见previous_date对象的active属性已被设为False，但提交插入操作后，数据库仅新增了行，原有激活行的active属性并未变更。

相关代码 #

models.py #

binded_maker = sessionmaker(bind=engine)

class ScenarioTrainingSchedule(Base):
    __tablename__ = "scenario_training_schedule"
    id = Column(Integer, primary_key=True)
    company_id = Column(Integer, nullable=False)
    scenario_id = Column(String, nullable=False)
    training_date = Column(Date, nullable=False)
    creation_date = Column(DateTime, nullable=False, server_default=func.now())
    active = Column(Boolean, nullable=False, default=True)

def update_previous_training_date_activity(mapper, connection, target):
    session = binded_maker.object_session(target)
    if session is not None:
        previous_date = (
            session.query(ScenarioTrainingSchedule)
            .filter(ScenarioTrainingSchedule.scenario_id == target.scenario_id)
            .filter(ScenarioTrainingSchedule.active == 1)
            # .order_by(TrainingDate.training_date.desc())
            .first()
        )

        if previous_date:
            attributes.set_committed_value(previous_date, "active", False)

database.py #

binded_maker = sessionmaker(bind=engine)
session = binded_maker()

插入逻辑服务文件 #

def insert_scenario_training_date(company_id, scenario_name, scenario_date):
    try:
        company_id = int(company_id)
        training_date = datetime.datetime.strptime(scenario_date, "%Y-%m-%d").date()
        training = ScenarioTrainingSchedule(
            company_id=company_id, scenario_id=scenario_name, training_date=training_date
        )

        session.add(training)
        session.commit()
    except Exception as e:
        session.rollback()
        logging.error(e)

1. attributes.set_committed_value用法错误：该方法用于设置对象的已提交状态值，会绕过SQLAlchemy的变更追踪机制——直接修改属性但不会将对象标记为“脏数据”，导致SQLAlchemy提交会话时无法检测到变更，不会生成对应的UPDATE语句。
2. 事件监听器未绑定：代码中仅定义了update_previous_training_date_activity函数，但未将其绑定到SQLAlchemy的事件（如before_insert）上，导致插入新数据时该函数根本未被触发执行。

1. 绑定事件监听器 #

在models.py中，将定义的函数绑定到ScenarioTrainingSchedule的before_insert事件，确保插入新行时触发逻辑：

from sqlalchemy import event

# 在ScenarioTrainingSchedule类定义后添加事件绑定
event.listen(ScenarioTrainingSchedule, 'before_insert', update_previous_training_date_activity)

2. 替换属性修改方式 #

放弃attributes.set_committed_value，直接赋值让SQLAlchemy追踪变更：

# 替换原代码中的set_committed_value行
if previous_date:
    previous_date.active = False

3. 可选：处理并发场景 #

如果存在并发插入的可能，建议添加行锁定避免冲突，同时补全排序逻辑确保获取最新的激活行：

previous_date = (
    session.query(ScenarioTrainingSchedule)
    .filter(ScenarioTrainingSchedule.scenario_id == target.scenario_id)
    .filter(ScenarioTrainingSchedule.active == True)
    .order_by(ScenarioTrainingSchedule.training_date.desc())
    .with_for_update()  # 锁定行，防止并发修改
    .first()
)

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