如何使用SQLAlchemy将在线ConsumerExpenditures数据库迁移至本地MySQL？

正确的调用顺序与实现代码 #

以下是完整可运行的迁移代码，包含结构同步和数据迁移的全流程：

import sqlalchemy
from sqlalchemy import MetaData, Table, text
from sqlalchemy.orm import sessionmaker

# --------------------------
# 1. 连接源数据库并反射所有表结构
# --------------------------
src_engine = sqlalchemy.create_engine(
    "mariadb+mariadbconnector://guest:relational@relational.fit.cvut.cz:3306/ConsumerExpenditures"
)
src_meta = MetaData()
# 反射源库所有表的结构到metadata对象
src_meta.reflect(bind=src_engine)
# 确认获取到的表名（这里对应EXPENDITURES, FAMILIES, GEOGRAPHIES三张表）
target_tables = list(src_meta.tables.keys())

# --------------------------
# 2. 初始化本地数据库连接，创建目标库
# --------------------------
# 先连接MySQL服务（不带具体库名）
local_engine_root = sqlalchemy.create_engine("mysql+pymysql://root:1234@localhost:3306")
with local_engine_root.connect() as conn:
    # 创建目标数据库（如果不存在）
    conn.execute(text("CREATE DATABASE IF NOT EXISTS ConsumerExpenditures"))
    conn.commit()

# 切换到目标数据库，创建专属操作引擎
local_engine = sqlalchemy.create_engine(
    "mysql+pymysql://root:1234@localhost:3306/ConsumerExpenditures"
)

# --------------------------
# 3. 在本地创建与源库一致的表结构
# --------------------------
# 直接利用源库的metadata在本地批量创建所有表
src_meta.create_all(bind=local_engine)

# --------------------------
# 4. 批量迁移表数据
# --------------------------
# 创建源库和本地库的会话对象
src_session = sessionmaker(bind=src_engine)()
local_session = sessionmaker(bind=local_engine)()

for table_name in target_tables:
    table = Table(table_name, src_meta, autoload_with=src_engine)
    # 分块读取源库数据（避免一次性加载过多数据导致内存溢出）
    src_data = src_session.query(table).yield_per(1000)
    # 批量插入本地库
    local_session.bulk_insert_mappings(table, [dict(row) for row in src_data])
    local_session.commit()
    print(f"完成表 {table_name} 的数据迁移")

# 关闭会话释放资源
src_session.close()
local_session.close()

你之前代码的问题点 #

1. 重复初始化MetaData()对象，覆盖了已经反射好的源库表结构信息
2. 创建本地数据库后，未切换到该库的专属引擎，后续操作默认在MySQL系统库下执行
3. 缺少数据迁移的核心逻辑，仅尝试创建单表但语法有误
4. 未处理事务提交，可能导致数据写入不完整

关键细节说明 #

• 反射元数据：src_meta.reflect(bind=src_engine)会一次性获取源库所有表的结构，无需单独指定表名
• 批量操作：yield_per(1000)用于分块读取数据，避免内存溢出；bulk_insert_mappings是SQLAlchemy提供的高效批量插入方法
• 事务控制：每次表迁移后提交事务，确保单表数据的一致性

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