使用SQLAlchemy将JSON插入MySQL：executemany等效方法及报错解决

错误原因 #

你遇到的TypeError: text() takes 1 positional argument but 2 were given，是因为text()函数仅接受SQL语句字符串作为参数，不能直接把参数列表val作为第二个参数传入。同时你的错误捕获代码里还有笔误：e.__dic__应为e.__dict__。

SQLAlchemy中对应mysql.connector的executemany()，核心是通过execute()方法传递参数列表来实现批量插入，下面是具体解决方案：

修复后的基础批量插入代码 #

修改参数传递方式，将val通过execute()的parameters参数传入，即可实现批量插入：

import json
from sqlalchemy import create_engine, text
from sqlalchemy import exc

# 请替换为你的数据库连接信息
# engine = create_engine('mysql+mysqlconnector://username:password@host:port/database')

data = []
with open('vehicle_data_usa_2014-2016.json', 'r', encoding="utf-8") as f:
    data = json.load(f)
try:
    sql = "INSERT INTO vehicle (carModelName, engineType, MPGhighway, MPGcity) VALUES (%s, %s, %s, %s)"
    val = [(x["model_id"], x["engine_type"], x["mpg_highway"], x["mpg_city"]) for x in data]
    with engine.begin() as conn:
        # 正确用法：text()仅传入SQL语句，参数通过parameters传递
        conn.execute(statement=text(sql), parameters=val)
except exc.SQLAlchemyError as e:
    err = str(e.__dict__['orig'])
    print('Error while connecting to MySQL', err)

更高效的批量插入方案（推荐） #

针对大量数据插入，SQLAlchemy支持生成合并式INSERT语句（INSERT ... VALUES (...), (...), ...），性能比普通批量插入更优，有两种实现方式：

方式1：使用命名参数与字典列表 #

import json
from sqlalchemy import create_engine, text
from sqlalchemy import exc

engine = create_engine('mysql+mysqlconnector://username:password@host:port/database')

data = []
with open('vehicle_data_usa_2014-2016.json', 'r', encoding="utf-8") as f:
    data = json.load(f)

try:
    sql = "INSERT INTO vehicle (carModelName, engineType, MPGhighway, MPGcity) VALUES (:carModelName, :engineType, :MPGhighway, :MPGcity)"
    # 转换为字典列表，适配命名参数
    val_dicts = [
        {
            "carModelName": x["model_id"],
            "engineType": x["engine_type"],
            "MPGhighway": x["mpg_highway"],
            "MPGcity": x["mpg_city"]
        }
        for x in data
    ]
    with engine.begin() as conn:
        conn.execute(text(sql), val_dicts)
except exc.SQLAlchemyError as e:
    err = str(e.__dict__['orig'])
    print('Error while connecting to MySQL', err)

方式2：使用SQLAlchemy的insert构造器 #

import json
from sqlalchemy import create_engine
from sqlalchemy.dialects.mysql import insert
from sqlalchemy import exc

engine = create_engine('mysql+mysqlconnector://username:password@host:port/database')

data = []
with open('vehicle_data_usa_2014-2016.json', 'r', encoding="utf-8") as f:
    data = json.load(f)

try:
    # 构造批量插入语句
    insert_stmt = insert("vehicle").values(
        [
            {
                "carModelName": x["model_id"],
                "engineType": x["engine_type"],
                "MPGhighway": x["mpg_highway"],
                "MPGcity": x["mpg_city"]
            }
            for x in data
        ]
    )
    with engine.begin() as conn:
        conn.execute(insert_stmt)
except exc.SQLAlchemyError as e:
    err = str(e.__dict__['orig'])
    print('Error while connecting to MySQL', err)

ORM风格批量插入（若定义了模型） #

如果你已经通过SQLAlchemy ORM定义了Vehicle模型，可以用更贴合ORM的方式批量插入：

import json
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from your_model_module import Vehicle  # 替换为你的模型文件路径

engine = create_engine('mysql+mysqlconnector://username:password@host:port/database')
Session = sessionmaker(bind=engine)

data = []
with open('vehicle_data_usa_2014-2016.json', 'r', encoding="utf-8") as f:
    data = json.load(f)

try:
    val_dicts = [
        {
            "carModelName": x["model_id"],
            "engineType": x["engine_type"],
            "MPGhighway": x["mpg_highway"],
            "MPGcity": x["mpg_city"]
        }
        for x in data
    ]
    with Session.begin() as session:
        # 直接插入字典映射，无需创建模型实例
        session.bulk_insert_mappings(Vehicle, val_dicts)
except exc.SQLAlchemyError as e:
    err = str(e.__dict__['orig'])
    print('Error while connecting to MySQL', err)

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