如何在Python中使用asyncio实现数据库读写分离路由？-Python教程-PHP中文网

SQLAlchemy异步读写分离需用两个独立异步引擎（主库写、从库读），通过contextvars实现请求级路由开关，禁止自动SQL解析判读写，事务内必须绑定单一引擎，否则触发InvalidRequestError。

如何在python中使用asyncio实现数据库读写分离路由？

asyncio下如何让SQLAlchemy自动路由读写请求

SQLAlchemy本身不支持异步读写分离，必须配合aiomysql或asyncpg等异步驱动，并手动控制连接池和执行路径。直接在create_async_engine里配多个URL没用——它只认一个URL，不会自动分流。

核心思路是：用两个独立的异步引擎（一个主库写，一个从库读），再通过自定义Session或上下文管理器决定走哪个。

写操作强制走write_engine，读操作默认走read_engine，但需显式标记“强一致性读”时也走主库
不能依赖session.execute()自动判断语句类型——SELECT可能带FOR UPDATE，仍是写操作
避免在事务中混用读写引擎，否则会报InvalidRequestError: This session is bound to multiple engines

如何用contextvars实现请求级读写路由开关

Web框架（如FastAPI）中，每个请求应能独立控制是否“强制读主库”。靠全局变量会串数据，必须用contextvars.ContextVar。

示例：

from contextvars import ContextVar
from sqlalchemy.ext.asyncio import AsyncSession
<h1>定义路由开关</h1><p>_read_from_master: ContextVar[bool] = ContextVar("read_from_master", default=False)</p><div class="aritcle_card flexRow artxards">
											<div class="artcardd flexRow">
												<a class="aritcle_card_img" rel="nofollow" href="/xiazai/skill9965" title="macOS UI 自动化技能包 —— 用 Python 控制你的 Mac"><img
														src="https://img.php.cn/upload/skill/001/246/273/6adaf20dbae7c507.jpg" alt="macOS UI 自动化技能包 —— 用 Python 控制你的 Mac" onerror="this.onerror='';this.src='/static/lhimages/moren/morentu.png'" ></a>
												<div class="aritcle_card_info flexColumn">
													<a rel="nofollow" href="/xiazai/skill9965" title="macOS UI 自动化技能包 —— 用 Python 控制你的 Mac" class="overflowclass">macOS UI 自动化技能包 —— 用 Python 控制你的 Mac</a>
													<p class="overflowclass">让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac，无需学习 AppleScript

系统要求
macOS 14+ (Sonoma)

Python 3.9+

需要开启：辅助功能 + 屏幕录制权限</p>
												</div>
												<a rel="nofollow" href="/xiazai/skill9965" title="macOS UI 自动化技能包 —— 用 Python 控制你的 Mac" class="aritcle_card_btn flexRow flexcenter"><b></b><span>下载</span>
												</a>
											</div>
										</div><p>def get_db_session() -> AsyncSession:
engine = write_engine if _read_from_master.get() else read_engine
return AsyncSession(engine)
</p>

在中间件中根据请求头（如X-Consistency: strong）调用_read_from_master.set(True)
注意ContextVar只在当前async task内有效，不能跨asyncio.create_task()传播，需手动copy
别在__aenter__里设值——AsyncSession实例化时就该确定引擎，不是执行时才选

为什么不能直接patch SQLAlchemy的execute方法做自动识别

有人想拦截session.execute()，解析SQL文本判断是否为SELECT再换引擎。这不可行：

SQL字符串可能被参数化、拼接、ORM动态生成，str(stmt).upper().startswith("SELECT")误判率极高
ORM级操作如session.get(User, 1)底层可能发SELECT ... FOR UPDATE，仍是写操作
asyncio下无法安全地在execute入口处切换引擎——AsyncConnection绑定的是具体engine，换engine等于换连接池，会破坏事务上下文
更严重的是，session.execute()返回的Result对象持有对原始连接的引用，中途换引擎会导致ResourceClosedError

读写分离后事务与连接池的实际约束

一旦启用读写分离，以下限制立刻生效，且无法绕过：

所有session.begin()开启的事务，后续所有操作（包括SELECT）都必须走同一个write_engine，否则报InvalidRequestError: This connection is not part of an active transaction
read_engine的连接池不能设pool_pre_ping=True（部分驱动不支持），需靠应用层心跳检测从库可用性
主从延迟导致的脏读，只能靠业务容忍或加SELECT ... FOR UPDATE回主库——但后者会降低读性能，得权衡
若用asyncpg，注意其min_size/max_size是按引擎独立计算的，两个引擎共用10个连接≠总共10个，而是各10个