如何使用sqlalchemy将查询结果导出为.sql文件？

核心逻辑 #

不需要找专门的导出封装API，可直接生成符合MySQL语法的标准SQL脚本，文件结构固定为三部分：

1. 头部配置字符集、关闭约束检查、开启事务，避免导入报错、提升导入速度
2. 按查询对应目标表，拼接批量INSERT语句写入查询结果
3. 尾部提交事务、恢复数据库默认配置

不要手动实现字符串转义逻辑，直接用SQLAlchemy自带的SQL编译器处理字面量，避免特殊字符（单引号、换行、emoji、反斜杠）触发语法错误。

可直接复用的代码 #

不需要安装额外依赖，基于你现有的SQLAlchemy环境即可运行：

from sqlalchemy import create_engine, select, literal_column
from sqlalchemy.sql import compiler
import datetime

# 替换为你自己的数据库连接串
engine = create_engine("mysql+pymysql://用户名:密码@数据库地址:端口/库名")

# 这里放你那15条select查询，格式为(导入目标表名, 查询语句)，关联查询结果也支持
query_list = [
    ("target_table_1", select(literal_column("*")).select_from("source_table_1")),
    ("target_table_2", select(literal_column("id,name,create_time")).select_from("source_table_2").where(literal_column("status") == 1)),
    # 剩余13条查询按相同格式追加即可
]

# 初始化SQL编译器，自动处理值转义
sql_compiler = compiler.SQLCompiler(engine.dialect, None)

def escape_value(val):
    """将Python类型值转换为MySQL可识别的合法字面量"""
    if val is None:
        return "NULL"
    elif isinstance(val, (int, float, bool)):
        return str(val).lower() if isinstance(val, bool) else str(val)
    elif isinstance(val, (datetime.datetime, datetime.date, datetime.time)):
        return f"'{val.isoformat()}'"
    else:
        return sql_compiler.render_literal_value(str(val), engine.dialect.STRING)

with open("export_result.sql", "w", encoding="utf-8") as f:
    # 写入文件头配置
    f.write("SET NAMES utf8mb4;\n")
    f.write("SET FOREIGN_KEY_CHECKS = 0;\n")
    f.write("SET UNIQUE_CHECKS = 0;\n")
    f.write("SET AUTOCOMMIT = 0;\n")
    f.write("START TRANSACTION;\n\n")

    with engine.connect() as conn:
        for table_name, query in query_list:
            result = conn.execute(query)
            columns = list(result.keys())
            rows = result.fetchall()
            if not rows:
                f.write(f"-- 表 {table_name} 查询结果为空，跳过\n\n")
                continue
            
            # 如果需要导入前清空目标表旧数据，取消下一行注释即可
            # f.write(f"TRUNCATE TABLE `{table_name}`;\n")

            # 拼接批量INSERT语句，导入效率远高于单条INSERT
            col_str = ", ".join([f"`{col}`" for col in columns])
            f.write(f"INSERT INTO `{table_name}` ({col_str}) VALUES \n")
            value_part = []
            for row in rows:
                row_val = ", ".join([escape_value(v) for v in row])
                value_part.append(f"({row_val})")
            f.write(",\n".join(value_part))
            f.write(";\n\n")
    
    # 写入文件尾，提交事务恢复配置
    f.write("COMMIT;\n")
    f.write("SET FOREIGN_KEY_CHECKS = 1;\n")
    f.write("SET UNIQUE_CHECKS = 1;\n")
    f.write("SET AUTOCOMMIT = 1;\n")

注意事项 #

• 如果单表查询结果超过1万条，建议把fetchall()改为每次拉取1000-5000条分批拼接INSERT，避免单条SQL过长超过MySQL的max_allowed_packet限制
• 导出完成的.sql文件直接执行mysql -u用户名 -p 目标库名 < export_result.sql即可完成导入，不需要额外格式转换
• 如果目标表有自增主键，查询结果里不要遗漏主键字段，否则会出现主键重复或者自增ID错乱的问题

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