如何用SQLAlchemy将表1数据插入到另一库的同结构空表2？

问题描述 #

我现有如下代码：

from sqlalchemy import create_engine
engine1 = create_engine('mysql://user:password@host1/schema', echo=True)
engine2 = create_engine('mysql://user:password@host2/schema')
connection1 = engine1.connect()
connection2 = engine2.connect()
table1 = connection1.execute("select * from table1")
table2 = connection2.execute("select * from table2")

我需要将table1中的所有数据插入到connection2下结构完全相同的空表table2中，请问该如何实现？我也可以将table1的数据转为字典后再插入table2。我从SQLAlchemy文档中了解到有相关方法，但示例均基于新建表使用new_table.insert()，不适用于已存在的表，特此求助。

解决方案 #

针对你的需求，我整理了几种实用的实现方式，你可以根据自己的场景选择：

方法1：用SQLAlchemy Core映射已有表并批量插入

这是最推荐的方式，不需要手动定义表结构，利用SQLAlchemy的元数据自动加载已有表的结构，然后批量插入数据：

from sqlalchemy import create_engine, Table, MetaData

# 初始化源和目标数据库连接
engine_source = create_engine('mysql://user:password@host1/schema', echo=True)
engine_target = create_engine('mysql://user:password@host2/schema')

with engine_source.connect() as conn_source, engine_target.connect() as conn_target:
    # 加载目标数据库中已存在的table2结构
    metadata_target = MetaData()
    target_table = Table('table2', metadata_target, autoload_with=engine_target)
    
    # 查询源表table1的所有数据，转为字典列表（字段名自动匹配）
    source_result = conn_source.execute("SELECT * FROM table1")
    source_rows = source_result.mappings().all()
    
    # 批量插入到目标表
    conn_target.execute(target_table.insert(), source_rows)
    # 提交事务（如果连接不是自动提交模式）
    conn_target.commit()

这个方法的优势是自动适配表结构，只要源表和目标表字段名一致，就不需要额外调整，而且批量插入的效率很高。

方法2：原生SQL批量插入（轻量快速）

如果不想用Core的表映射，也可以直接构造原生SQL语句来批量插入，适合简单场景：

from sqlalchemy import create_engine

engine_source = create_engine('mysql://user:password@host1/schema', echo=True)
engine_target = create_engine('mysql://user:password@host2/schema')

with engine_source.connect() as conn_source, engine_target.connect() as conn_target:
    # 获取源表所有数据和字段名
    source_result = conn_source.execute("SELECT * FROM table1")
    source_rows = source_result.fetchall()
    column_names = source_result.keys()
    
    if source_rows:
        # 构造批量插入SQL
        insert_sql = f"""
            INSERT INTO table2 ({', '.join(column_names)})
            VALUES ({', '.join(['%s'] * len(column_names))})
        """
        # 执行批量插入
        conn_target.execute(insert_sql, source_rows)
        conn_target.commit()

注意：这种方式要求源表和目标表的字段顺序和数量完全一致，否则会出现数据错位的问题。

方法3：ORM Session批量插入（如果已有模型类）

如果你的项目中已经定义了对应table1/table2的ORM模型类，可以用Session的批量方法来操作：

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
# 假设你已经在models.py中定义了对应表的模型类TableModel
from models import TableModel

engine_source = create_engine('mysql://user:password@host1/schema', echo=True)
engine_target = create_engine('mysql://user:password@host2/schema')

# 创建两个数据库的Session类
SessionSource = sessionmaker(bind=engine_source)
SessionTarget = sessionmaker(bind=engine_target)

with SessionSource() as session_source, SessionTarget() as session_target:
    # 查询源表所有数据
    source_data = session_source.query(TableModel).all()
    # 批量保存到目标数据库
    session_target.bulk_save_objects(source_data)
    session_target.commit()

提示：bulk_save_objects不会触发ORM的生命周期事件（比如before_insert钩子），但插入效率比逐个添加高很多，如果需要触发事件，可以改用add_all()方法。

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