SQLAlchemy 2.0+ 的 update() 不支持 executemany 批量更新，传入多行参数会降级为单条循环执行；应优先使用 bulk_update_mappings()（需主键已知）或方言特定的 upsert（如 insert().on_conflict_do_update()）。
SQLAlchemy 2.0+ 的 update() 不再支持 executemany 风格批量更新
Python 3.12 下用 SQLAlchemy 2.0+（如 2.0.30+）做批量更新，不能直接对一堆字典调用 session.execute(update(...), [...])——这会触发单条 SQL 执行 N 次，性能极差。根本原因是 Update 构造对象默认不启用底层的 executemany 优化，即使传入多行参数，它也会降级为循环执行。
真正有效的批量更新必须绕过 ORM 层的 session.bulk_update_mappings() 或直接走 Core 的 execute() + 原生批量语句。但要注意：bulk_update_mappings() 要求主键已知且不能触发事件或验证，而原生语句则需手动处理字段映射和类型转换。
-
bulk_update_mappings()仅适用于模型已有主键值、且你信任数据完整性的情况 - 若依赖数据库自增 ID 或需要 upsert 逻辑（如不存在则插入），得改用
insert(...).on_conflict_do_update()（PostgreSQL）或REPLACE INTO/INSERT ... ON DUPLICATE KEY UPDATE（MySQL） - SQLite 不支持
ON CONFLICT的完整语法，得用INSERT OR REPLACE+ 全字段覆盖，或拆成SELECT+UPDATE两步
用 bulk_update_mappings() 更新已有主键的爬虫记录
这是最常用也最稳妥的方案，适合爬虫入库后二次修正字段（比如补全 status、updated_at），前提是每条记录的主键（如 id 或 url_hash）已在 DB 中存在且你手上有对应值。
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
<h1>假设 Model 定义为 class Article(Base):</h1><h1><strong>tablename</strong> = "articles"</h1><h1>id = Column(Integer, primary_key=True)</h1><h1>title = Column(String)</h1><h1>status = Column(String)</h1><p>engine = create_engine("sqlite:///crawler.db")
Session = sessionmaker(bind=engine)
session = Session()</p><h1>爬虫采集后整理出待更新的数据列表（必须含主键字段）</h1><p>updates = [
{"id": 101, "title": "新标题A", "status": "parsed"},
{"id": 102, "title": "新标题B", "status": "parsed"},
{"id": 103, "title": "新标题C", "status": "failed"},
]</p><h1>注意：字段名必须与模型属性名一致，不能是数据库列名（如不能写 "article_title"）</h1><p>session.bulk_update_mappings(Article, updates)
session.commit()
- 必须确保
updates中每个 dict 都包含完整的主键字段，否则会抛KeyError或静默跳过 - 不会调用
__init__、@validates、before_update等钩子，也不校验类型，纯数据灌入 - MySQL/PostgreSQL 下实际执行的是单条
UPDATE ... WHERE id = ?的批量绑定，不是一条 SQL 更新多行——但比循环session.query().filter().update()快 5–10 倍
用 Core 的 update() + executemany=True 实现真批量（需数据库驱动支持）
如果你用的是较新版本的 psycopg（v3.1+）或 pymysql（v1.1+），可以强制开启底层批量执行。但注意：SQLAlchemy 默认禁用该行为，必须显式配置引擎并构造语句。
图片提示词生成器？不止如此。 马甲系统 —— 把脑海中的画面，翻译成AI能理解的专业表达。 用得越多，它越懂你：首次需要多问几句确认方向，用久了几乎一说就懂。 用得越多，它越快：缓存机制让后续对话越来越省。 RAG进化：成功案例持续入库，越跑越聪明。 输入「新手指南」查看完整功能介绍
立即学习“Python免费学习笔记（深入）”；
from sqlalchemy import update, text
from sqlalchemy.dialects.postgresql import insert
<h1>PostgreSQL 示例（推荐用 insert().on_conflict_do_update 更安全）</h1><p>stmt = (
update(Article)
.where(Article.id == text("data.id"))
.values(
title=text("data.title"),
status=text("data.status"),
updated_at=text("NOW()")
)
)</p><h1>注意：这不是标准写法，而是配合 VALUES (...) FROM (VALUES ...) 的变体</h1><h1>真正高效的写法依赖数据库方言，通常不如 bulk_update_mappings 直观</h1><p>- SQLite 和旧版 MySQL 驱动不支持
executemany对UPDATE的优化，强行传多行参数仍会循环执行 - PostgreSQL 可用
insert().on_conflict_do_update()替代，一条语句完成 upsert，且自动批量绑定 - 别在
update().values(...)里直接写 Python 变量，否则 SQL 注入；要用text()或参数化占位符
爬虫场景下更推荐的替代路径：先 bulk_insert_mappings() 再去重更新
多数爬虫入库本质是“有则更新、无则插入”，与其纠结批量 update，不如用带冲突处理的插入——既避免主键缺失问题，又天然支持批量。SQLAlchemy 2.0+ 对各数据库的 upsert 支持已较成熟。
# PostgreSQL
stmt = insert(Article).values(updates)
stmt = stmt.on_conflict_do_update(
index_elements=["url_hash"], # 唯一键或索引列
set_=dict(
title=stmt.excluded.title,
status=stmt.excluded.status,
updated_at=func.now()
)
)
session.execute(stmt)
session.commit()
- MySQL 用
insert(...).on_duplicate_key_update(...) - SQLite 用
insert(...).on_conflict_replace()（需 3.24+）或退化为INSERT OR REPLACE - 关键点：必须提前在数据库中为用于判断“是否存在”的字段（如
url_hash）建唯一索引，否则 upsert 无效
实际批量更新时最容易被忽略的，是数据库连接是否启用了 fast_executemany=True（SQL Server）、use_batch_mode=True（PyMySQL）或 prepare_statement=True（psycopg）。这些开关不在 SQLAlchemy 层控制，而在底层驱动初始化时设置，不配就永远达不到理论吞吐量。
