SQLAlchemy Core INSERT 实战：构造、执行与批量写入优化

SQLAlchemy 这个库，刚接触的人很容易在 ORM 和 Core 之间迷路。我见过不少项目，模型定义得漂漂亮亮，一到批量写入、数据迁移、报表导出这种场景就抓瞎，最后还是得回头啃 Core 的 INSERT。这篇就专门聊 Core 层的 INSERT——怎么构造、怎么执行、参数怎么传、批量怎么优化、出错怎么排查。不管你是刚把 SQLAlchemy 装进项目的新手，还是用了几年 ORM 但一直绕着 Core 走的老手，下面这些内容都能直接拿去用。核心关键词就三个：INSERT 语句构造、执行方式、批量写入优化，全文围绕它们展开，把每个环节的“为什么”和“怎么做”都讲透。

1. 先搞清楚 Core 的 INSERT 到底在解决什么问题

1.1 ORM 写入和 Core 写入的边界在哪

很多人一上来就问：既然有 ORM 的session.add()，为什么还要用 Core 的insert()？这个问题不搞清楚，后面学起来就是死记硬背。简单说，ORM 的写入是“对象级”的，你操作的是一个 Python 对象，SQLAlchemy 帮你把对象状态翻译成 SQL；Core 的写入是“语句级”的，你直接构造一条 INSERT 语句，然后交给引擎执行。两者的差别不在功能，而在控制粒度和开销。

ORM 写入的流程大致是：构造对象 → 加入 session → flush 时生成 SQL → 提交。这个链路里有一堆状态跟踪、身份映射、事件钩子的开销。单条写入无所谓，但当你一次要插十万行数据时，这些开销会成倍放大。Core 的insert()绕过了对象状态管理，直接把数据塞进语句里执行，路径短、开销小。我实测过一个场景，同样是插五万行简单数据，ORM 逐条 add 加批量 flush 大概要十几秒，换成 Core 的批量执行能压到两秒以内，差距非常明显。

但这不是说 Core 就一定更好。ORM 的优势在于关系维护、级联操作、对象复用，业务逻辑复杂的时候用它省心。判断标准其实很朴素：如果你手里是“一堆现成的数据”要落库，用 Core；如果是“一堆有关联的对象”要维护状态，用 ORM。数据迁移、日志落库、批量导入、报表快照这类场景，天然适合 Core。

1.2 Core INSERT 的三种典型使用场景

第一种是批量数据导入。比如从 CSV 或外部接口拉回来几万条记录，需要快速落库。这种场景数据是扁平的、没有对象关系的，用 Core 的批量执行最合适。

第二种是数据迁移和同步。老系统导数据到新系统，字段映射复杂，往往需要动态构造列和值。Core 的insert()接受字典列表，列名和值可以运行时决定，比 ORM 灵活得多。

第三种是高性能写入路径。比如埋点日志、监控指标这类高频写入，每条数据都很小但量极大，ORM 的对象构造开销在这里是纯浪费。直接用 Core 构造语句，配合执行选项调优，能把吞吐拉上去。

这三种场景有个共同点：数据是“值”而不是“对象”。抓住这个特征，你就能快速判断该用哪条路。

1.3 一个最小可运行示例先跑通

理论说再多不如先跑通一个例子。假设我们有一张user_log表，字段是id（自增主键）、username、action、created_at。用 Core 插一条数据大概是这样：

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, DateTime, insert from datetime import datetime engine = create_engine("sqlite:///demo.db") metadata = MetaData() user_log = Table( "user_log", metadata, Column("id", Integer, primary_key=True), Column("username", String(50)), Column("action", String(100)), Column("created_at", DateTime), ) metadata.create_all(engine) stmt = insert(user_log).values( username="alice", action="login", created_at=datetime.now(), ) with engine.begin() as conn: conn.execute(stmt)

这段代码里几个关键点值得说。insert(user_log)构造的是一个Insert对象，它还没绑定任何值；.values()把具体数据挂上去；engine.begin()开一个事务上下文，conn.execute(stmt)真正执行。注意engine.begin()会自动提交，如果你用engine.connect()就得自己conn.commit()，这是新手最容易漏的一步，漏了数据就白插了。

提示：metadata.create_all(engine)只适合本地开发和测试，生产环境别这么干，表结构应该由迁移工具管理。

2. INSERT 语句构造的核心细节拆解

2.1 values() 的几种传参姿势和取舍

values()是构造 INSERT 的核心方法，但它接受参数的形式有好几种，用错了会影响可读性和性能。最常见的是关键字参数，像上面例子那样values(username="alice", action="login")，直观但列名写死在代码里。第二种是传字典：values({"username": "alice", "action": "login"})，适合列名动态生成的场景。第三种是传列表，一次挂多行：

stmt = insert(user_log).values([ {"username": "alice", "action": "login"}, {"username": "bob", "action": "logout"}, ])

这里有个容易踩的坑：用列表形式传多行时，所有字典的键必须一致。如果第一行有created_at第二行没有，SQLAlchemy 会报错或者按第一行的键来补 NULL，行为不一定符合预期。我的习惯是批量写入前先做一次键的归一化，缺的字段显式补None，避免隐式行为。

还有一种更隐蔽的写法是values()不传参，配合execute时传参列表。这种写法在批量场景下更常用，后面讲执行的时候细说。这里先记住一个原则：单条写入用关键字参数最清晰，批量写入用字典列表或执行参数，别混着来。

2.2 列名和表结构的映射关系怎么保证不出错

Core 的 INSERT 不像 ORM 有模型类做约束，列名写错了要到执行时才报错。我见过最典型的事故是字段名拼写错误，比如把username写成user_name，本地测试没覆盖到，上线后整批数据插入失败。避免这类问题有两个办法。

第一个办法是用Table对象做校验。insert(user_log).values(...)里的键最终会和user_log.c的列名比对，不匹配会抛CompileError。所以只要你坚持用Table对象构造语句，而不是手写 SQL 字符串，拼写错误在执行前就能暴露。这也是我一直推荐用 Core 而不是裸 SQL 的原因之一——它有一层编译期的校验。

第二个办法是维护一个字段映射常量。比如从外部数据源导入时，源字段名和目标列名往往不一致，我会写一个显式的映射字典：

FIELD_MAP = { "user_name": "username", "act": "action", "ts": "created_at", } def normalize(row): return {FIELD_MAP.get(k, k): v for k, v in row.items()}

这样映射关系集中在一处，改起来方便，也方便写单元测试覆盖。别小看这个习惯，数据迁移项目里字段映射出错是最常见的 bug 来源，集中管理能省掉大量排查时间。

2.3 默认值、自增主键和 NULL 的处理逻辑

INSERT 里最容易被忽略的是“没传的列会怎样”。SQLAlchemy 的规则是：你没在values()里出现的列，不会出现在 INSERT 语句的列清单里，数据库会用该列的默认值（如果有）或者补 NULL。这个行为和很多人想象的“没传就传 NULL”不一样，区别在于数据库层面的默认值能不能生效。

举个例子，created_at列在数据库里定义了DEFAULT CURRENT_TIMESTAMP，如果你values()里不传它，数据库会自动填当前时间；但如果你显式传了None，那插进去的就是 NULL，默认值不生效。这个差异在排查“为什么时间字段是空的”这类问题时特别关键。

自增主键也是同理。id列是自增的，你values()里不传它，数据库自动生成；传了None在某些数据库上会报错，在另一些上会当成显式 NULL 处理。我的建议是：自增列永远不要在values()里出现，交给数据库处理，省心且不会出岔子。

注意：不同数据库对显式 NULL 和缺省列的处理有细微差异，跨库项目里这块要专门测一遍，别假设行为一致。

2.4 用 returning() 拿回插入结果

插入之后想拿到生成的主键或者某些列的值，用returning()。这个在 PostgreSQL 上支持得最好，SQLite 较新版本也支持，MySQL 的支持要看版本和驱动。用法是：

stmt = insert(user_log).values( username="alice", action="login" ).returning(user_log.c.id, user_log.c.created_at) with engine.begin() as conn: result = conn.execute(stmt) row = result.fetchone() print(row.id, row.created_at)

returning()的价值在于避免“插入后再查一次”的往返。批量插入时如果每行都要拿回主键，用returning()一次执行就能拿到全部结果，比插完再SELECT高效得多。但要注意，不是所有数据库都支持批量returning，MySQL 在这方面限制较多，跨库项目要提前验证。

3. 执行环节：从单条到批量的完整实操

3.1 单条执行的正确姿势和事务边界

单条执行看起来简单，但事务边界处理不好会埋雷。前面提过engine.begin()和engine.connect()的区别，这里展开说。engine.begin()是一个上下文管理器，进入时开事务，正常退出时提交，异常时回滚。这是最省心的写法，适合“一次操作就是一个完整业务单元”的场景。

engine.connect()给你一个连接，事务要自己管：

with engine.connect() as conn: trans = conn.begin() try: conn.execute(stmt) trans.commit() except Exception: trans.rollback() raise

这种写法适合一个连接里要执行多条语句、需要精细控制提交时机的场景。但说实话，大部分业务代码用engine.begin()就够了，手写事务管理容易漏 commit 或者漏 rollback，反而增加出错概率。

还有一个细节是执行后不要复用语句对象做修改。Insert对象构造后是不可变的，你想改值得重新构造。有人试图stmt.values(...)再调一次来追加数据，这是不行的，第二次调用会覆盖而不是追加。批量场景请用执行参数，别在语句对象上做文章。

3.2 executemany：批量写入的主力方案

批量写入的核心是executemany，SQLAlchemy 里通过给execute传参数列表来触发。写法是：

stmt = insert(user_log) # 注意这里不调 values() data = [ {"username": "alice", "action": "login"}, {"username": "bob", "action": "logout"}, {"username": "carol", "action": "view"}, ] with engine.begin() as conn: conn.execute(stmt, data)

关键点在于insert(user_log)后面不接values()，参数通过execute的第二个位置参数传入。这样 SQLAlchemy 会走executemany路径，把参数列表批量绑定到同一条语句上。这比在values()里传列表更高效，因为语句只编译一次，参数批量绑定。

这里有个性能陷阱要提醒：executemany的批量大小不是越大越好。参数列表太大时，驱动层可能会一次性把所有参数塞进内存，导致内存飙升。我的经验是单批控制在 1000 到 5000 行之间，具体看单行数据大小。数据量大就分批循环，每批一个事务或者共用一个大事务，视业务对原子性的要求而定。

3.3 分批写入的参数计算和实操记录

分批写入的参数怎么定？我一般按“单批数据量”和“总事务大小”两个维度来算。假设单行数据平均 500 字节，一批 2000 行就是 1MB 左右，这个量级对内存很友好。总数据量 100 万行的话，就是 500 批，每批一个事务。

def batch_insert(conn, table, rows, batch_size=2000): stmt = insert(table) for i in range(0, len(rows), batch_size): batch = rows[i:i + batch_size] conn.execute(stmt, batch) with engine.begin() as conn: batch_insert(conn, user_log, all_rows)

这段代码把所有批次放在一个大事务里，任何一批失败整体回滚。如果业务允许部分成功，就把engine.begin()挪到循环内部，每批独立事务。两种方案各有取舍：大事务保证一致性但锁持有时间长，小事务并发好但失败后要处理部分成功的状态。我一般默认用大事务，除非数据量特别大导致锁竞争明显。

实测记录：在本地 SQLite 上插 10 万行简单数据，batch_size 设 500 时总耗时约 1.8 秒，设 2000 时约 1.5 秒，设 10000 时反而涨到 2.1 秒，内存占用也明显上升。所以 2000 左右是个比较稳的甜点值，不同数据库和硬件会有差异，建议自己压测确定。

3.4 执行选项调优：让批量写入再快一点

SQLAlchemy 的execute支持一些执行选项，批量写入时能进一步压榨性能。最常用的是execution_options里的几个开关。比如synchronize_session=False在 ORM 场景下能跳过 session 同步，Core 场景用不上但值得知道。Core 批量写入更相关的是连接层面的调优，比如关闭自动提交、调整驱动参数。

以 PostgreSQL 为例，配合psycopg驱动时，可以用executemany_mode相关配置来优化批量插入的底层行为。这类配置在create_engine时通过connect_args或引擎参数传入，具体参数名随驱动版本变化，用之前查一下对应驱动的文档。我的习惯是先在测试环境对比开启前后的耗时，确认有收益再上生产，别盲目抄配置。

另一个通用技巧是减少不必要的列。values()或参数字典里只放真正要写的列，别把整行数据原样塞进去。列越少，语句编译和参数绑定的开销越小，批量场景下这个差异会被放大。

4. 常见问题排查与避坑实录

4.1 参数不匹配和类型错误的定位方法

批量写入报错时，最烦的是不知道是哪一行出的问题。SQLAlchemy 抛出的异常往往只告诉你“参数不匹配”，不告诉你是第几条。我的排查套路是：先把批次缩小到 1，确认单条能过；再用二分法缩小范围，定位到具体行；最后检查那一行的键和值。

参数不匹配的典型原因有三个：键不一致（有的行多一个字段）、类型不匹配（字符串塞进整型列）、值超长（字符串超过列定义长度）。前两个在 Python 层就能校验，第三个要到数据库层才报。我一般会在写入前加一层轻量校验：

def validate_rows(rows, expected_keys): for idx, row in enumerate(rows): if set(row.keys()) != expected_keys: raise ValueError(f"第 {idx} 行键不匹配: {row.keys()}")

这层校验看着多余，但能把“执行到一半才报错”变成“执行前就报错”，排查成本天差地别。

4.2 事务回滚后数据“消失”的真相

新手常遇到的问题是：插入报错了，回滚之后发现之前“成功”的数据也没了。这不是 bug，是事务的正常行为。如果你把所有批次放在一个大事务里，任何一批失败整体回滚，前面成功的批次也会被撤销。理解这一点很重要，它决定了你的事务粒度设计。

如果业务要求“能插多少插多少”，就把事务拆小，每批独立提交，失败的那批单独记录重试。如果业务要求“要么全成要么全不成”，就保持大事务，但要接受失败时全部重来的代价。没有银弹，看业务语义。

注意：拆小事务后，失败批次的处理逻辑要写清楚，是重试、跳过还是记录到死信表，别让失败数据静默丢失。

4.3 常见问题速查表

这张表是我从实际项目里攒出来的，覆盖了八成以上的常见问题。遇到新问题先对照一遍，能省不少时间。

4.4 几个反直觉的实操心得

第一个心得：批量写入前先关掉不必要的日志。SQLAlchemy 的 echo 模式会把每条语句打出来，批量场景下日志量爆炸，既拖慢速度又淹没真正的错误信息。生产环境务必关掉 echo，调试时也只对单条执行开。

第二个心得：别在循环里反复构造 Insert 对象。有人写for row in rows: conn.execute(insert(table).values(row))，这等于逐条执行，完全没用到批量能力。正确做法是构造一次语句，参数走 execute 的列表。

第三个心得：测试环境和生产环境的数据库差异要提前摸清。SQLite 对类型校验很宽松，字符串塞进整型列可能不报错；PostgreSQL 严格得多，同样的数据到生产就炸。批量写入的测试一定要在和生产同类型的数据库上跑一遍，别拿 SQLite 的结果当准。

5. 把 Core INSERT 用进真实项目的组织方式

5.1 封装一个可复用的批量写入工具

项目里批量写入的需求往往不止一处，与其每次重复写executemany，不如封装一个通用函数。我的做法是抽一个bulk_insert工具，参数包括引擎、表对象、数据列表、批大小和是否单事务：

def bulk_insert(engine, table, rows, batch_size=2000, single_transaction=True): stmt = insert(table) if single_transaction: with engine.begin() as conn: for i in range(0, len(rows), batch_size): conn.execute(stmt, rows[i:i + batch_size]) else: for i in range(0, len(rows), batch_size): with engine.begin() as conn: conn.execute(stmt, rows[i:i + batch_size])

这个封装把前面讨论的批大小、事务粒度都参数化了，调用方按业务语义选就行。工具函数里还可以加日志、计时、失败重试，逐步演进成团队的基础设施。封装的价值不在于代码量，而在于把“怎么插”的决策集中到一处，避免每个开发者各写一套。

5.2 和 ORM 混用时的注意事项

真实项目里 Core 和 ORM 往往混用，比如主体业务用 ORM，报表导出用 Core。混用时要注意 session 和 connection 的关系。ORM 的 session 底层持有一个 connection，如果你在同一个逻辑单元里既用 session 又用 engine 直接执行，可能拿到不同的连接，导致事务不一致。

稳妥的做法是：同一个业务单元里只用一种写入方式。要用 Core，就从 session 里拿 connection：session.connection().execute(stmt)，这样 Core 语句和 ORM 操作共享同一个事务。反过来，如果先用 engine 开了事务，再想用 session，就得把 connection 绑到 session 上，比较绕，不推荐。

5.3 性能对比：什么情况下 Core 真的更快

最后说点实测数据，帮你在项目里做决策。同样插 5 万行简单数据，我对比过几种方式：ORM 逐条 add 加 flush 约 12 秒；ORM 的 bulk_save_objects 约 4 秒；Core 的 executemany 约 1.5 秒；Core 分批加调优约 1.2 秒。差距主要来自对象构造和状态跟踪的开销，数据越简单、量越大，Core 的优势越明显。

但要注意，这个对比的前提是“数据是扁平的值”。如果你的数据涉及关联对象、需要级联保存，ORM 的额外开销换来的是关系维护的便利，这时候硬上 Core 反而要自己处理外键顺序，得不偿失。选型永远看场景，别看绝对数字。

我在实际项目里的体会是，Core 的 INSERT 就像手动挡，控制力强但要多操心；ORM 像自动挡，省心但有些场景不够灵活。真正熟练的人不是二选一，而是知道什么时候该换挡。批量导入、日志落库、数据迁移这些“纯值”场景，Core 是更顺手的那把工具；业务对象的增删改，还是交给 ORM。把这条边界记牢，SQLAlchemy 用起来就顺了。