SQLAlchemy批量操作如何获取bulk_insert/update_mappings影响行数

首先可以明确，session.bulk_insert_mappings() 和 session.bulk_update_mappings() 本身不会直接返回操作影响的行数，但可以通过两种常见方案实现需求：

方案1：改用 execute 执行批量语句直接获取 rowcount #

这是准确度最高、实现最简单的方案，和你熟悉的 connection.execute() 逻辑完全一致：

• 对于批量插入，直接调用表对象的 insert 方法执行批量映射，返回结果直接读取 rowcount 属性即可
• 对于批量更新，同理使用表对象的 update 方法配合 bindparam 绑定参数执行
  示例代码如下：

from sqlalchemy.orm import Session
from sqlalchemy import bindparam
from your_project import User # 替换为你自己的模型类

# 批量插入示例
insert_data = [
    {"username": "user1", "age": 18},
    {"username": "user2", "age": 20}
]
insert_result = session.execute(
    User.__table__.insert(),
    insert_data
)
insert_affected_rows = insert_result.rowcount
session.commit()

# 批量更新示例
update_data = [
    {"id": 1, "age": 19},
    {"id": 2, "age": 21}
]
update_result = session.execute(
    User.__table__.update()
    .where(User.id == bindparam("id"))
    .values(age=bindparam("age")),
    update_data
)
update_affected_rows = update_result.rowcount
session.commit()

*注意：无冲突场景下，批量插入的rowcount和你传入的映射列表长度一致，如果存在INSERT IGNORE或者ON DUPLICATE KEY UPDATE这类逻辑，rowcount会返回数据库实际插入/更新的行数，比直接统计列表长度更准确。

方案2：保留 bulk 方法写法，通过事件监听统计行数 #

如果你不想改动现有使用bulk_insert_mappings/bulk_update_mappings的业务代码，可以通过监听SQLAlchemy的会话事件来统计行数：

from sqlalchemy import event
from sqlalchemy.orm import Session

# 全局统计变量，也可以绑定到session对象上避免全局污染
bulk_insert_count = 0
bulk_update_count = 0

@event.listens_for(Session, "do_bulk_insert_mappings")
def handle_bulk_insert(mapper, connection, mappings, *args, **kwargs):
    global bulk_insert_count
    # 无插入忽略场景下，映射数量等于实际插入行数
    bulk_insert_count += len(mappings)

@event.listens_for(Session, "do_bulk_update_mappings", retval=True)
def handle_bulk_update(mapper, connection, mappings, *args, **kwargs):
    global bulk_update_count
    # 执行更新获取实际影响行数，返回结果阻止默认执行避免重复操作
    result = connection.execute(
        mapper._bulk_update_statement,
        mappings
    )
    bulk_update_count += result.rowcount
    return result

*注意：该方案的插入统计仅在无数据库层面插入忽略/冲突跳过的场景下准确，如果有这类逻辑，建议还是改用方案1直接拿执行结果的rowcount。

额外说明 #

所有基于rowcount的统计都依赖你使用的数据库驱动对该属性的正确实现，部分数据库（比如旧版本SQLite）在批量操作场景下可能返回不准确的数值，业务对行数准确性要求高的话，建议批量操作后单独执行count查询做二次校验。

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