Flask-SQLAlchemy不支持自动读写分离，因其默认仅连接SQLALCHEMY_DATABASE_URI指定的单库，binds仅管理模型绑定而非语句路由；SELECT/INSERT等操作不会按类型自动分发至从库或主库，必须手动通过using_bind()、get_engine()或自定义DBRouter显式指定连接。
为什么不能直接用 Flask-SQLAlchemy 自动做读写分离
Flask-SQLAlchemy 本身不提供读写分离能力，db.session 默认只连一个 SQLALCHEMY_DATABASE_URI。即使你手动配置多个 URI，它也不会自动把 SELECT 发到从库、INSERT/UPDATE/DELETE 发到主库——所有操作都走同一个引擎。
常见错误是以为加个 binds 就能分流，结果发现 query 还是走主库，或者事务里混用主从导致数据不一致。
-
binds只控制模型绑定哪个数据库，不控制语句类型路由 - 手动切换
session.bind容易漏掉嵌套查询或关系加载（比如.join()或lazy='joined'） - 事务必须全程在主库执行，一旦在从库开了 session 再切回主库，会报
InvalidRequestError: Can't reconnect until invalid transaction is rolled back
用 SQLAlchemy 的 create_engine + 路由中间件实现最简读写分离
核心思路：不依赖 ORM 的 session 绑定机制，而是自己封装一个 DBRouter 类，在每次执行前判断 SQL 类型，动态选主库或从库的 Engine。
关键点在于拦截的是 Connection.execute()，而不是 ORM 层的 query——这样能覆盖原生 SQL、ORM 查询、even session.execute()。
- 主库用
create_engine(..., pool_pre_ping=True)避免连接失效 - 从库建议加
pool_recycle=3600和connect_args={'read_timeout': 5}防卡死 - 路由逻辑必须区分
SELECT（含SELECT COUNT、SELECT EXISTS）和写操作，但不要用正则匹配全 SQL，而应检查compiled.dialect.name和compiled.is_select
class DBRouter:
def __init__(self, master_url, slave_urls):
self.master = create_engine(master_url, pool_pre_ping=True)
self.slaves = [create_engine(u, pool_recycle=3600) for u in slave_urls]
<pre class="brush:php;toolbar:false;">def get_engine(self, clauseelement=None):
if clauseelement is None:
return self.master
compiled = clauseelement.compile()
if hasattr(compiled, 'is_select') and compiled.is_select:
return random.choice(self.slaves)
return self.master
在 Flask 请求周期里安全注入主从连接
不能全局复用 engine，也不能在模块顶层创建 connection——得绑定到请求上下文。Flask 的 g 是合适载体，但要注意：如果用了 before_request 初始化连接，必须确保 after_request 或 teardown_request 显式关闭，否则连接泄漏。
更稳妥的做法是按需获取连接，并用 contextlib 管理生命周期：
- 读操作：用
router.get_engine().connect()，查完立刻.close() - 写操作：必须用主库
begin()启动事务，且整个事务块内所有语句都走同一 connection - 避免在模板里调用查询（如
{{ user.posts }}），这会触发延迟加载，可能跑到从库去查关联数据，而外层事务在主库，造成不一致
示例中，get_user_by_id() 应该明确指定用主库（即使只是 SELECT），因为后续可能跟更新；而 list_articles() 才走从库：
@app.route('/articles')
def list_articles():
conn = db_router.get_engine(text('SELECT 1')).connect() # 显式走从库
rows = conn.execute(text('SELECT * FROM articles LIMIT 20')).fetchall()
conn.close()
return render_template('list.html', articles=rows)
主从延迟下哪些场景必须强制走主库
MySQL 主从复制有毫秒到秒级延迟，不是所有 SELECT 都能扔给从库。以下情况一旦走从库，大概率出 bug：
- 刚插入/更新完立刻查（比如注册后跳转个人页，查刚写的
user记录） - 涉及
FOR UPDATE或锁读（从库不支持写锁，会报错ERROR 1290 (HY000): The MySQL server is running with the --read-only option） - 跨库 JOIN 且主从分属不同实例（从库没权限或没同步对应库）
- 使用了主库特有的函数或变量，如
LAST_INSERT_ID()、@var用户变量
这类逻辑没法靠自动路由识别，必须人工标注。建议封装一个 force_master=True 参数，或约定某些视图函数名带 _master 后缀，路由时优先匹配。
真正难处理的是“读己之写”（read-your-writes）场景——用户自己提交的数据要立刻可见。这时候要么加主库读兜底，要么引入缓存层（如 Redis）暂存刚写入的 key，查不到再 fallback 到主库。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
