Flask-SQLAlchemy实现PostgreSQL UPSERT解决area_id唯一键冲突问题

问题原因说明 #

• 全局编译INSERT语句的方案不符合PostgreSQL语法要求：PostgreSQL标准UPSERT语法顺序为INSERT ... ON CONFLICT ... [RETURNING]，之前的编译逻辑将ON CONFLICT放在RETURNING之后触发语法错误，且全局修改所有插入逻辑风险极高，不推荐使用
• TypeError: unhashable type: 'dict'报错是因为调用on_conflict_do_update时参数传递错误：要么是指定唯一约束的index_elements/constraint参数传入了字典类型，要么是更新字段的set_参数格式不符合要求

正确实现步骤 #

1. 确认Area模型定义 #

首先确保你的模型中area_id已设置唯一约束，示例模型如下：

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

class Area(db.Model):
    __tablename__ = 'area'
    id = db.Column(db.Integer, primary_key=True)
    area_id = db.Column(db.String(64), unique=True, nullable=False) # 唯一键约束
    # 其他字段示例
    area_name = db.Column(db.String(128), nullable=False)
    province = db.Column(db.String(64))
    city = db.Column(db.String(64))
    vote_count = db.Column(db.Integer)

2. 构造UPSERT语句 #

使用sqlalchemy.dialects.postgresql.insert构造插入语句，冲突时基于area_id唯一约束更新所有字段，无需手动枚举所有字段：

from sqlalchemy.dialects.postgresql import insert

# 此处替换为你解析CSV得到的实际数据列表，格式为字典数组
csv_data = [{"area_id": "110000", "area_name": "北京市", "province": "北京市", "vote_count": 1234}]

# 构造基础插入语句
insert_stmt = insert(Area).values(csv_data)

# 构造冲突更新规则：匹配area_id唯一约束，自动更新所有非主键、非唯一键字段
update_dict = {c.name: c for c in insert_stmt.excluded if c.name not in ['id', 'area_id']}
upsert_stmt = insert_stmt.on_conflict_do_update(
    index_elements=['area_id'], # 指定用于冲突匹配的唯一约束字段
    set_=update_dict
)

# 执行语句并提交事务
db.session.execute(upsert_stmt)
db.session.commit()

3. 特殊场景适配 #

如果需要保留原始数据的部分字段不被CSV新数据覆盖，调整update_dict的生成逻辑即可，比如保留首次入库的create_time字段：

update_dict = {
    c.name: c for c in insert_stmt.excluded 
    if c.name not in ['id', 'area_id', 'create_time']
}

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