如何让Cloud SQL异步连接池在main函数外供其他函数调用？

要让连接池在main函数之外供其他函数调用，核心是管理好Connector和连接池的生命周期，确保它们在程序运行期间保持可用，而非局限在main的async with块内。以下是具体实现方案：

1. 定义全局存储变量 #

在模块级别声明变量，用于保存Connector实例和连接池，确保全局可访问：

import asyncio
import asyncpg
import sqlalchemy
from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
from google.cloud.sql.connector import Connector

# 全局变量：存储连接池和Connector实例
_db_pool: AsyncEngine | None = None
_connector: Connector | None = None

2. 编写异步初始化函数 #

在程序启动时调用该函数，完成Connector和连接池的初始化：

async def init_db():
    global _db_pool, _connector
    loop = asyncio.get_running_loop()
    
    # 初始化Connector（不再用async with，避免自动关闭）
    _connector = Connector(loop=loop)
    
    # 定义连接创建逻辑
    async def getconn() -> asyncpg.Connection:
        return await _connector.connect_async(
            "project:region:instance",  # 替换为你的Cloud SQL实例连接名
            "asyncpg",
            user="my-user",
            password="my-password",
            db="my-db-name"
            # 可添加额外数据库驱动参数
        )
    
    # 创建异步连接池
    _db_pool = create_async_engine(
        "postgresql+asyncpg://",
        async_creator=getconn,
    )

3. 提供连接池访问入口 #

写一个全局函数，供其他模块/函数获取已初始化的连接池：

def get_db_pool() -> AsyncEngine:
    if _db_pool is None:
        raise RuntimeError("数据库连接池未初始化，请先调用init_db()")
    return _db_pool

4. 编写资源清理函数 #

在程序退出时调用，正确关闭连接池和Connector，避免资源泄漏：

async def close_db():
    global _db_pool, _connector
    # 先关闭连接池，再关闭Connector
    if _db_pool is not None:
        await _db_pool.dispose()
        _db_pool = None
    if _connector is not None:
        await _connector.close_async()
        _connector = None

5. 其他函数调用示例 #

任何需要使用数据库的异步函数，都可以通过get_db_pool()获取连接池：

async def fetch_current_time():
    pool = get_db_pool()
    async with pool.connect() as conn:
        result = await conn.execute(sqlalchemy.text("SELECT NOW()"))
        print("当前数据库时间:", result.scalar())

6. 程序入口修改 #

在main中完成初始化、业务逻辑执行和资源清理：

async def main():
    try:
        # 初始化数据库连接池
        await init_db()
        # 调用其他使用连接池的函数
        await fetch_current_time()
    finally:
        # 确保程序退出时清理资源
        await close_db()

if __name__ == "__main__":
    asyncio.run(main())

关键注意事项 #

• 生命周期绑定：Connector的生命周期必须覆盖连接池的使用周期，不能提前关闭。
• 异步上下文要求：所有数据库操作必须在异步函数中执行，不能在同步函数中直接调用。
• 框架适配：如果使用FastAPI、Starlette等异步Web框架，可以通过lifespan事件来执行init_db和close_db，实现启动初始化、退出清理的自动执行。

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