单个连接池能否存储多数据库连接？是否需创建多个连接池？

针对多数据库切换场景，并非只能创建多个独立连接池，以下是几种更高效的实现方案：

1. 动态绑定引擎（最常用方案） #

SQLAlchemy支持在Session中动态切换绑定的数据库引擎，每个引擎自带独立连接池，无需手动管理连接的创建与销毁。

from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker

# 为每个数据库创建独立引擎（自带连接池）
db1_engine = create_engine("postgresql://user:pass@db1:5432/db1", pool_size=10, max_overflow=20)
db2_engine = create_engine("mysql+pymysql://user:pass@db2:3306/db2", pool_size=10, max_overflow=20)

# 创建不指定默认绑定的Session工厂
Session = scoped_session(sessionmaker())

# 切换到db1执行操作
session = Session(bind=db1_engine)
session.query(DB1Model).filter_by(id=1).first()
session.commit()

# 切换到db2执行操作
session.bind = db2_engine
session.query(DB2Model).filter_by(name="test").all()
session.commit()

这种方式本质上是管理多个连接池，但通过Session统一入口，避免了手动维护连接的繁琐，连接池的复用、回收完全由SQLAlchemy自动处理。

2. 自定义连接池路由 #

如果需要更细粒度的控制（比如根据请求上下文动态选择数据库），可以自定义连接池子类，实现连接的路由逻辑。

from sqlalchemy.pool import Pool
from sqlalchemy import create_engine
import threading

# 线程本地变量存储当前要连接的数据库标识
local_storage = threading.local()

def set_current_db(db_key):
    local_storage.db_key = db_key

def get_current_db():
    return getattr(local_storage, "db_key", "db1")

class RoutingPool(Pool):
    def __init__(self, pool_map, **kwargs):
        self.pool_map = pool_map
        super().__init__(**kwargs)
    
    def _create_connection(self):
        # 根据当前上下文选择对应连接池
        current_db = get_current_db()
        return self.pool_map[current_db]._create_connection()

# 创建各数据库的连接池
db1_pool = create_engine("postgresql://user:pass@db1:5432/db1").pool
db2_pool = create_engine("mysql+pymysql://user:pass@db2:3306/db2").pool

# 初始化路由池，封装多个连接池为统一入口
routing_pool = RoutingPool({"db1": db1_pool, "db2": db2_pool})

# 创建绑定路由池的引擎
engine = create_engine("postgresql://user:pass@db1:5432/db1", pool=routing_pool)

# 使用时通过上下文切换数据库
set_current_db("db2")
with engine.connect() as conn:
    result = conn.execute("SELECT * FROM test_table")

3. ShardedSession（分库分表场景） #

如果是分库分表的业务场景，SQLAlchemy的ShardedSession可以根据分片键自动路由到对应数据库的连接池。

from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.shard import ShardedSession

def shard_chooser(mapper, instance, clause=None):
    # 根据实例的分片键选择数据库
    return "db1" if instance.user_id % 2 == 0 else "db2"

def id_chooser(query, ident):
    # 根据ID选择对应的分片库
    user_id = ident[0]
    return ["db1"] if user_id % 2 == 0 else ["db2"]

def query_chooser(query):
    # 根据查询条件选择分片库
    for child in query.whereclause.get_children():
        if hasattr(child, 'left') and child.left.name == 'user_id':
            user_id = child.right.value
            return ["db1"] if user_id % 2 == 0 else ["db2"]
    return ["db1", "db2"]

# 创建多数据库引擎
engines = {
    "db1": create_engine("postgresql://user:pass@db1:5432/db1"),
    "db2": create_engine("postgresql://user:pass@db2:5432/db2")
}

# 配置分片Session
Session = sessionmaker(class_=ShardedSession)
Session.configure(
    shard_chooser=shard_chooser,
    id_chooser=id_chooser,
    query_chooser=query_chooser,
    shards=engines
)

# 使用时自动路由
session = Session()
session.add(User(user_id=1, name="Alice"))  # 自动路由到db2
session.commit()

总结 #

每个数据库本质上需要独立的连接池（因为连接参数、会话隔离要求不同），但无需手动创建和维护多个独立的池实例：

• 简单多库切换优先用动态绑定引擎，实现成本低，兼容大部分场景；
• 自定义路由需求用连接池路由，实现灵活的上下文感知切换；
• 分库分表场景用ShardedSession，原生支持分片逻辑与连接池管理。

内容的提问来源于stack exchange，提问作者Sai Vamshi