SQLAlchemy更新数据时出现sqlalchemy.exc.IntegrityError报错如何解决？

问题根因 #

你遇到的报错是SQLAlchemy一对多关系的默认关联处理逻辑和数据库非空约束冲突导致的：
当你直接给new_forecast.forecasts赋值新列表时，如果new_forecast原来已有关联的Forecast记录，SQLAlchemy默认会先把这些旧记录的job_id设为NULL，断开它们和new_forecast的关联，再把新列表里的Forecast记录的job_id更新为new_forecast的ID。而你的revenue表的job_id字段有NOT NULL约束，所以第一步设NULL的操作直接触发了完整性错误。

另外你代码里还有个潜在的笔误：Job类的表名是job，但Forecast的外键写的是db.ForeignKey('forecast_job.id')，二者不匹配，建议先修正为db.ForeignKey('job.id')。

解决方案 #

第一步：调整relationship的级联规则 #

给Job类的forecasts关系加上delete-orphan级联，这条规则的作用是：当Forecast记录从父对象（Job）的关联列表中移除时，直接删除这条Forecast记录，而不是尝试把外键设为NULL，刚好符合你要覆盖旧数据的需求。

class Job(db.Model):
    __tablename__ = 'job'
    
    id = db.Column(db.Integer(), primary_key=True)
    job_code = db.Column(db.String(4), nullable=True)
    job_description = db.Column(db.String(200), nullable=False)    
    # 新增delete-orphan级联，可选添加backref简化双向关联
    forecasts = db.relationship('Forecast', order_by="Forecast.time", cascade="all, delete-orphan", backref="job")

第二步：调整合并逻辑 #

避免删除旧Job时级联删除已经迁移过去的Forecast记录，调整后的代码如下：

# Open session to run operation
engine = create_engine(current_app.config["SQLALCHEMY_DATABASE_URI"])
Session = sessionmaker(bind = engine)  
mer_session = Session() 

# Select jobs
old_forecast = mer_session.query(Job).filter(Job.id == manual_id).first()
new_forecast = mer_session.query(Job).filter(Job.id == auto_id).first()

# 把要迁移的Forecast转成独立列表，脱离和旧Job的关联
migrated_forecasts = list(old_forecast.forecasts)
# 直接赋值，SQLAlchemy会自动删除new_forecast原有的Forecast记录
# 同时自动更新迁移过来的Forecast的job_id为new_forecast的ID
new_forecast.forecasts = migrated_forecasts

# 删除旧Job
mer_session.delete(old_forecast)

#Commit changes
mer_session.commit()

如果已经给relationship加了backref="job"，上述代码可以直接正常运行：当Forecast被添加到new_forecast的关联列表时，会自动从old_forecast的关联列表中移除，删除old_forecast时不会误删已经迁移的记录。

如果没有加backref，只需要在删除old_forecast前加一行清空旧关联的代码即可：

old_forecast.forecasts = []

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