如何在SQLAlchemy的upsert操作中分别统计插入和更新的记录数量

你可以利用PostgreSQL内置的系统隐藏字段xmax实现插入/更新记录的区分，不需要新增自定义字段：

• 新插入的行未经过任何更新/删除操作，xmax值固定为0
• 被ON CONFLICT DO UPDATE逻辑更新的行，xmax会被赋值为当前执行事务的ID，值必然大于0

修改后代码示例 #

from sqlalchemy import literal_column
from sqlalchemy.dialects.postgresql import insert
from sqlalchemy import BigInteger
from flask_sqlalchemy import SQLAlchemy

# 注意：原代码直接将SQLAlchemy类别名设为db属于常见用法错误，正确用法是先实例化
db = SQLAlchemy()

class PetModel(db.Model):
    __tablename__ = "pets"

    id = db.Column(BigInteger, primary_key=True)
    data = db.Column(db.String(64), nullable=False, unique=True)

    def as_dict(self):
        return {
            "id": self.id,
            "data": self.data,
        }

pets = [PetModel(id=1, data="Dog"), PetModel(id=2, data="Cat")]

insert_statement = insert(PetModel).values([_.as_dict() for _ in pets])
# 在upsert语句中加入returning，新增is_inserted标记位
upsert_statement = insert_statement.on_conflict_do_update(
    constraint="pet_pkey",
    set_={"data": insert_statement.excluded.data},
).returning(
    PetModel.__table__,
    (literal_column("xmax") == 0).label("is_inserted")
)

result = db.session.execute(upsert_statement)
db.session.commit()

# 统计插入和更新数量
rows = result.all()
insert_count = sum(1 for row in rows if row.is_inserted)
update_count = len(rows) - insert_count

print(f"插入记录数：{insert_count}，更新记录数：{update_count}")

说明 #

• 该方案仅适用于PostgreSQL数据库，xmax是PostgreSQL独有的系统事务字段，无需在模型定义或表结构中新增任何字段
• 统计结果的总和和result.rowcount返回的总操作数完全一致

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