直接用SQLAlchemy管理爬虫数据可行，但需规避自动commit/flush：默认Session非线程安全，串行add+commit吞吐低、事务回滚导致前序成功数据丢失；应改用bulk_insert_mappings或原生upsert语句，按worker隔离Session，并始终用with上下文管理生命周期。
直接用 SQLAlchemy 管理爬虫数据是可行的，但必须绕开“自动 commit + 自动 flush”这个坑——否则高并发写入时会卡死、丢数据、报 InvalidRequestError: This Session's transaction has been rolled back。
为什么爬虫场景下不能直接用 session.add() + session.commit() 串行写入？
爬虫通常批量解析、异步获取、多线程/协程入库，而默认 Session 不是线程安全的，且每次 commit() 都触发完整事务提交+连接归还，吞吐量极低。更麻烦的是，如果某条数据违反唯一约束（比如重复抓取的 URL），整个事务回滚，前面成功的记录也丢了。
- 避免单次 commit 写多条：改用
session.bulk_insert_mappings()或session.execute(insert(...).prefix_with("IGNORE"))（MySQL） - 每个 worker 线程/协程必须持有独立
Session实例，不要复用全局 session 对象 - 显式控制连接生命周期：用
with Session(engine) as session:，别靠垃圾回收关连接
如何定义兼顾爬虫字段变更和查询效率的 Model？
爬虫数据结构常变动（新增字段、调整类型）、字段多但常用查询条件少，硬套严格 ORM 映射反而拖慢开发。推荐“核心字段 ORM + 扩展字段 JSON”混合模式。
- 主键、URL、状态、抓取时间等高频查询字段走标准列定义，加
Index（如Index('ix_article_url', 'url')） - 正文、作者、标签等易变/不定长字段统一存进
JSON类型（PostgreSQL）或Text+ 应用层json.loads()（SQLite/MySQL） - 避免用
PickleType：反序列化风险高，且无法在 SQL 层过滤
示例片段：
立即学习“Python免费学习笔记（深入）”；
from sqlalchemy import Column, Integer, String, Text, DateTime, Index, JSON
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime
<p>Base = declarative_base()</p><div class="aritcle_card flexRow">
<div class="artcardd flexRow">
<a class="aritcle_card_img" href="/xiazai/skill11254" title="Python Dependency Candidate Fetcher 2"><img
src="https://img.php.cn/upload/skill/001/246/273/6adaf20d79f23518.jpg" alt="Python Dependency Candidate Fetcher 2" ></a>
<div class="aritcle_card_info flexColumn">
<a href="/xiazai/skill11254" title="Python Dependency Candidate Fetcher 2">Python Dependency Candidate Fetcher 2</a>
<p>根据包的能力以及 Jupyter 环境中的导入语句，从候选列表中挑选最适合用户表格工作流的 Python 包。</p>
</div>
<a href="/xiazai/skill11254" title="Python Dependency Candidate Fetcher 2" class="aritcle_card_btn flexRow flexcenter"><b></b><span>下载</span> </a>
</div>
</div><p>class Article(Base):
<strong>tablename</strong> = 'articles'
id = Column(Integer, primary_key=True)
url = Column(String(2048), nullable=False, index=True)
status = Column(String(20), default='pending')
fetched_at = Column(DateTime, default=datetime.utcnow)
raw_data = Column(JSON) # PostgreSQL；MySQL 可换为 Text怎么安全地去重插入（upsert）而不爆锁？
“先查再插”在并发下必然产生重复，而原生 INSERT ... ON CONFLICT（PostgreSQL）或 INSERT IGNORE（MySQL）又没法直接对接 ORM 的 add() 流程——得绕过 ORM，直连 Core。
- PostgreSQL：用
insert(Article).on_conflict_do_nothing(index_elements=['url']) - MySQL：用
insert(Article).prefix_with("INSERT IGNORE") - SQLite：不支持 upsert 原语，需用
INSERT OR IGNORE INTO ...字符串执行，或改用merge+ 主键匹配（性能差） - 无论哪种，都必须确保冲突字段有唯一索引，否则 upsert 无效
关键代码：
from sqlalchemy.dialects.postgresql import insert
<p>stmt = insert(Article).values(url="<a href="https://www.php.cn/link/b05edd78c294dcf6d960190bf5bde635">https://www.php.cn/link/b05edd78c294dcf6d960190bf5bde635</a>", raw_data={"title": "test"})
stmt = stmt.on_conflict_do_nothing(index_elements=['url'])
session.execute(stmt)
session.commit()session.close() 和 engine.dispose() 到底什么时候调？
很多人以为爬虫跑完调一次 engine.dispose() 就完事，结果下次启动时报 “Connection is closed”。根本原因是：engine 是连接池管理者，dispose() 会干掉所有空闲连接，但正在用的连接（比如没 close 的 session）不受控，下次取连接时可能拿到已失效句柄。
- 每个请求/任务单元结束时，必须调
session.close()（或用with上下文） -
engine.dispose()只应在整个爬虫进程退出前调一次，比如atexit.register(lambda: engine.dispose()) - 长时间运行的爬虫服务（如 APScheduler 定时抓取），完全不需要手动 dispose，连接池自己会清理超时连接
最稳妥的写法就是始终用上下文管理：
with Session(engine) as session:
session.execute(...)
session.commit() # 或不 commit，依赖上下文自动 rollback连接泄漏比数据错乱更隐蔽——它不会立刻报错，而是慢慢耗尽数据库连接数，最后整个服务静默拒绝新请求。
