SQLAlchemy多数据库绑定简单吗

本文目录导读：

1. 核心概念
2. 场景一：最简单的多数据库绑定（仅查询不同库）
3. 场景二：读/写分离（主从）
4. 场景三：复杂的动态路由（多租户）

SQLAlchemy 的多数据库绑定功能相对简单,但具体取决于你的使用场景。

可以这样概括：

• 对于“读/写分离”或“主从复制”场景：非常简单直接。
• 对于“多库（多租户、不同业务库）”场景：中等难度,需要一点配置和路由逻辑。

核心概念

你需要理解两个关键部分：

1. 引擎 (Engine)：绑定到一个具体的数据库。
2. 会话 (Session)：通过 binds 参数，决定每个 Model 使用哪个引擎。

最简单的多数据库绑定（仅查询不同库）

假设你有两个数据库：users_db 和 orders_db。

配置引擎

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy.ext.declarative import declarative_base
# 创建两个引擎
engine_users = create_engine("postgresql://user:pass@host:5432/users_db")
engine_orders = create_engine("postgresql://user:pass@host:5432/orders_db")
# 创建基类
Base = declarative_base()

定义模型并绑定

关键在于 __bind_key__ 属性（这是 Flask-SQLAlchemy 的用法，纯 SQLAlchemy 有不同方式，但逻辑类似）。在纯 SQLAlchemy 中，你需要手动在会话中指定 binds。

# 模型定义 (使用普通的 Base)
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String)
class Order(Base):
    __tablename__ = "orders"
    id = Column(Integer, primary_key=True)
    amount = Column(Float)

创建带绑定信息的 Session

这是核心步骤：

# 创建 Session 时，使用 binds 参数
SessionLocal = sessionmaker(
    binds={
        User: engine_users,   # User 模型的所有操作 -> engine_users
        Order: engine_orders  # Order 模型的所有操作 -> engine_orders
    }
)
# 使用
session = SessionLocal()
# 以下查询自动路由到正确的数据库
users = session.query(User).all()   # 自动走 engine_users
orders = session.query(Order).all() # 自动走 engine_orders
session.close()

这种方式非常简单，你只需在创建 sessionmaker 时配置一次 binds 字典，后续所有 CRUD 操作都是透明的。

读/写分离（主从）

读/写分离本质上是多绑定的一种特例。

• User 模型：写操作 -> master 引擎；读操作 -> slave 引擎。
• 仅靠简单的 binds 字典无法实现自动读写分离（因为 SQLAlchemy 不会自动区分当前 SQL 是 SELECT 还是 INSERT）。

你需要自己写一个路由（Routing Session）或使用第三方库。

示例（简单的读写分离 Session）：

from sqlalchemy.orm import Session as BaseSession
class RoutingSession(BaseSession):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        # 假设你传入了 master/slave 引擎
        self._master_engine = kwargs.pop('master', None)
        self._slave_engines = kwargs.pop('slaves', [])
    def get_bind(self, mapper=None, clause=None, **kw):
        # 核心逻辑：判断是读还是写
        if self._is_write(clause): # 需要你自己实现判断逻辑
            return self._master_engine
        else:
            import random
            return random.choice(self._slave_engines)  # 负载均衡
    def _is_write(self, clause):
        # 简化判断：如果包含 INSERT, UPDATE, DELETE 则为写
        # 实际中推荐使用 inspect(clause).is_select
        return not clause.is_select if clause else False

难度：中等。 需要理解 SQLAlchemy 的 Session 内部 get_bind 方法并覆盖它。

复杂的动态路由（多租户）

每个租户有不同的数据库连接。

class DynamicRoutingSession(BaseSession):
    def get_bind(self, mapper=None, clause=None, **kw):
        # 假设你有个函数 get_db_for_tenant()
        # 通过请求上下文获取商户ID
        tenant_id = some_global_request_context.get("tenant_id")
        engine = get_engine_by_tenant(tenant_id)  # 从连接池获取
        return engine

难度：中等偏高。 需要处理好上下文线程安全（例如用 Flask 的 g 或 Starlette 的 contextvar）。

推荐方案：

• 简单多库：直接用 sessionmaker(binds=...),这是最简单的原生方案。
• 读写分离：使用 sqlalchemy-rider 或 sqlalchemy-mate 等第三方工具,比自己造轮子安全。
• 多租户：强烈建议使用 SQLAlchemy 2.0 + AsyncSession，或者直接使用 Flask-SQLAlchemy 3.x，它对多绑定支持更好（__bind_key__）。

一句话结论： 如果你的需求是“不同模型在不同库”，那确实非常简单（5行代码配置），如果是动态路由,那就不简单了。