如何在单个asyncpg连接池中切换多个数据库？

核心思路 #

基于你现有的单例连接池，我们通过以下方式满足需求：

• 利用PostgreSQL的CONNECT命令在已建立的连接上直接切换目标数据库，无需重建连接，最大化降低系统负载
• 为连接池添加异步上下文管理器，简化连接的获取、切换与释放流程

修改后的完整代码实现 #

import asyncpg
from contextlib import asynccontextmanager
from your_config_module import cfg  # 替换为你的配置模块路径

class ConnectionPool:
    _instance = None

    def __new__(cls, *args, **kwargs):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

    async def init(self):
        # 初始化连接池时不指定具体数据库，连接到PostgreSQL默认库（如postgres）
        self.pool = await asyncpg.create_pool(
            dsn=f"postgres://{cfg.DB_USER}:{cfg.DB_PASSWORD}@{cfg.DB_HOST}:{cfg.DB_PORT}"
        )

    @asynccontextmanager
    async def connection(self, db_name):
        """
        异步上下文管理器：获取连接并切换到指定数据库，自动处理连接释放
        :param db_name: 目标数据库名称
        """
        conn = await self.pool.acquire()
        try:
            # 在已获取的连接上切换到目标数据库
            await conn.execute(f"CONNECT {db_name};")
            yield conn
        finally:
            # 无论是否发生异常，都将连接释放回池
            await self.pool.release(conn)

使用示例 #

async def example_usage():
    # 初始化单例连接池（全局只需调用一次）
    db_pool = ConnectionPool()
    await db_pool.init()

    # 切换到db1执行查询操作
    async with db_pool.connection("db1") as conn:
        result = await conn.fetch("SELECT id, name FROM users;")
        print("db1查询结果:", result)

    # 切换到db2执行插入操作
    async with db_pool.connection("db2") as conn:
        await conn.execute("INSERT INTO logs(content) VALUES ($1);", "用户登录日志")
        print("db2插入完成")

关键注意事项 #

• 权限要求：确保数据库用户拥有访问目标数据库的CONNECT权限
• 连接复用：连接池中的连接会被复用，但每次通过connection()方法获取时都会重新切换到指定数据库，避免数据交叉污染
• 异常处理：如果目标数据库不存在、用户无权限等，CONNECT命令会抛出asyncpg.PostgresError，可在业务代码中捕获处理
• 负载优化：这种方式复用同一连接池的连接，避免为每个数据库单独创建连接池，大幅降低系统资源消耗

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