优化Flush：开启cascade='all'时实现单表多行INSERT批量插入

我想优化当前操作，让每个表（比如trans、scripts）使用单条INSERT INTO ... VALUES (...), (...)批量插入语句，而非多条单行INSERT。但因为用了cascade="all"，我仅将父对象（cfg）添加到会话，所有关联子对象（Trans、Script）会在session.flush()时自动添加。这种情况下没法用bulk_save_objects，因为它不支持关系关联，也不能正确处理主键分配。

请问有没有办法在flush期间拦截或按表分组生成的INSERT语句（无需数据库干预），手动构建每个表的批量插入（多行）？

Base = declarative_base()

class Config(Base):
    __tablename__ = 'configs'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    trans = relationship("Trans", back_populates="config", cascade="all")
    scripts = relationship("Script", back_populates="config", cascade="all")

class Trans(Base):
    __tablename__ = 'trans'
    id = Column(Integer, primary_key=True)
    config_id = Column(Integer, ForeignKey("configs.id"))
    name = Column(String)
    config = relationship("Config", back_populates="trans")

class Script(Base):
    __tablename__ = 'scripts'
    id = Column(Integer, primary_key=True)
    config_id = Column(Integer, ForeignKey("configs.id"))
    name = Column(String)
    config = relationship("Config", back_populates="scripts")


engine = create_engine("sqlite:///:memory:", echo=True)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine, autoflush=True)
session = Session()

cfg = Config(name="MyConfig")
cfg.trans.append(Trans(name="T1"))
cfg.trans.append(Trans(name="T2"))
cfg.scripts.append(Script(name="S1"))
cfg.scripts.append(Script(name="S2"))

# 关联子对象会通过cascade自动处理
session.add(cfg)
session.flush()

方法1：手动拆分操作，先插父对象再批量插子对象 #

放弃级联自动处理，手动控制插入顺序，直接用批量插入API：

1. 先添加父对象并flush，拿到生成的主键
2. 给所有子对象手动设置外键值
3. 按表调用bulk_save_objects批量插入

示例代码：

# 插入父对象，获取主键
session.add(cfg)
session.flush()  # 此时cfg.id已生成

# 批量插入Trans
trans_list = [Trans(name="T1", config_id=cfg.id), Trans(name="T2", config_id=cfg.id)]
session.bulk_save_objects(trans_list)

# 批量插入Script
script_list = [Script(name="S1", config_id=cfg.id), Script(name="S2", config_id=cfg.id)]
session.bulk_save_objects(script_list)

session.commit()

这个方法简单直接，性能提升明显，唯一缺点是需要手动维护外键关联，无法依赖级联特性。

方法2：用SQLAlchemy事件监听拦截并合并INSERT语句 #

通过before_execute和after_execute事件，缓存同表的单行INSERT语句，最终合并成批量INSERT执行。

示例代码：

from sqlalchemy import event
from sqlalchemy.engine import Engine

# 缓存每个表的待插入字段和数据
insert_cache = {}

@event.listens_for(Engine, "before_execute")
def before_execute(conn, clauseelement, multiparams, params):
    # 只处理INSERT语句
    if hasattr(clauseelement, 'insert'):
        table_name = clauseelement.table.name
        columns = [col.name for col in clauseelement.columns]
        values = multiparams[0]

        if table_name not in insert_cache:
            insert_cache[table_name] = {'columns': columns, 'values': []}
        
        # 字段匹配则缓存数据，跳过当前单行插入
        if insert_cache[table_name]['columns'] == columns:
            insert_cache[table_name]['values'].append(values)
            return clauseelement, [], {}
        # 字段不匹配则先执行缓存的批量插入，再处理当前语句
        else:
            execute_bulk(conn, table_name)
            insert_cache[table_name] = {'columns': columns, 'values': [values]}
    return clauseelement, multiparams, params

@event.listens_for(Engine, "after_execute")
def after_execute(conn, clauseelement, multiparams, params, result):
    # 所有语句执行完毕后，清空缓存并执行剩余批量插入
    for table_name in list(insert_cache.keys()):
        execute_bulk(conn, table_name)
    insert_cache.clear()
    return result

def execute_bulk(conn, table_name):
    cache = insert_cache[table_name]
    if not cache['values']:
        return
    # 构建批量INSERT语句
    placeholders = ', '.join(['(' + ', '.join(['%s']*len(row)) + ')' for row in cache['values']])
    insert_sql = f"INSERT INTO {table_name} ({', '.join(cache['columns'])}) VALUES {placeholders}"
    # 扁平化参数并执行
    flat_params = [item for row in cache['values'] for item in row]
    conn.execute(insert_sql, flat_params)

注意：这个方法需要自己处理主键回填（比如数据库自增主键，要把生成的ID同步到ORM对象），还要适配不同数据库的语法差异（比如部分数据库对批量插入的行数有限制）。

方法3：用bulk_insert_mappings以字典形式批量插入 #

如果可以接受用字典代替ORM对象，这个方法性能比bulk_save_objects更优：

session.add(cfg)
session.flush()

# 批量插入Trans
trans_mappings = [{'name': 'T1', 'config_id': cfg.id}, {'name': 'T2', 'config_id': cfg.id}]
session.bulk_insert_mappings(Trans, trans_mappings)

# 批量插入Script
script_mappings = [{'name': 'S1', 'config_id': cfg.id}, {'name': 'S2', 'config_id': cfg.id}]
session.bulk_insert_mappings(Script, script_mappings)

session.commit()

和方法1类似，需要手动处理外键，但避免了创建ORM对象的开销。

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