使用 SQLAlchemy Core 替代 ORM 批量插入
如果 ORM 的 session.add_all() 或 bulk_insert_mappings 失效，尝试使用 Core 的 insert().values() 方法：
python

复制

下载
from sqlalchemy import create_engine, table, column, Integer,
String

engine
= create_engine("dm+pyodbc://user:password@host:port/dbname")
conn
= engine.connect()

定义表结构

my_table
= table("my_table",
column
("id", Integer),
column
("name", String)
)

批量插入数据

data
= [
{"id": 1, "name": "Alice"},
{"id": 2, "name": "Bob"},
{"id": 3, "name": "Charlie"}
]

conn
.execute(my_table.insert().values(data))
conn
.commit()
3. 分批插入数据
若一次性插入大量数据导致超时或内存问题，可分批处理：

python
batch_size = 1000 # 每批插入 1000 条
for i in range(0, len(data), batch_size):
batch = data[i:i+batch_size]
conn.execute(my_table.insert().values(batch))
conn.commit() # 每批提交一次
4. 使用 executemany 直接操作游标
绕过 SQLAlchemy 的抽象层，直接使用底层驱动的 executemany 方法：

python
from sqlalchemy import create_engine

engine = create_engine("dm+pyodbc://user:password@host:port/dbname")
conn = engine.raw_connection() # 获取原生连接
cursor = conn.cursor()

执行批量插入

sql = "INSERT INTO my_table (id, name) VALUES (?, ?)"
data = [(1, "Alice"), (2, "Bob"), (3, "Charlie")]
cursor.executemany(sql, data)

conn.commit()
cursor.close()