FastAPI如何在子路由间共享单一数据库连接池

项目目录结构 #

-/main.py
-/db.py
--/routers/
---device.py
---/operations/
----backup.py

包含main.py、db.py文件，以及routers目录（内含device.py和operations子目录，operations下有backup.py）。

现有代码 #

main.py #

from starlette.responses import RedirectResponse
from fastapi import FastAPI, Request

# Import routers
from routers import device

app = FastAPI()

@app.get("/")
async def root(request: Request):
    return RedirectResponse(url=f"{request.url}docs/")

# Load routers
app.include_router(device.router)

device.py #

### Device API Router
from fastapi import APIRouter

# Import custom classes
from . import db
from routers.operations import backup

# Parse the DB config
db_pool = db() # This initializes a pool of MySQL connections using the db.py file

router = APIRouter()

# Import subrouters
router.include_router(backup.router)

核心需求 #

1. 让backup.py等子路由文件复用同一个数据库连接池，避免重复创建多个连接池，最终保持一个10-20连接的单一连接池供所有路由共享。
2. 按操作拆分API到不同文件（如/device/backup/、/device/network/），保持代码结构清晰。

方案一：将连接池封装为单例模块（最直接易用） #

通过修改db.py实现单例模式，确保全局只初始化一次连接池，所有路由文件直接导入使用即可。

1. 重构db.py为单例模式 #

# db.py
class DBPool:
    _instance = None
    
    def __new__(cls):
        # 确保只创建一个实例
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            # 初始化MySQL连接池，设置连接数范围
            cls._instance.pool = cls._init_connection_pool()
        return cls._instance
    
    @staticmethod
    def _init_connection_pool():
        # 替换为你的实际连接池初始化代码（示例使用asyncmy）
        import asyncmy
        return asyncmy.pool.create_pool(
            host="localhost",
            user="your_db_user",
            password="your_db_pass",
            db="device_management",
            minsize=10,
            maxsize=20
        )

# 创建全局单例实例，所有导入都复用这个对象
db_pool = DBPool()

2. 修改device.py和backup.py使用单例连接池 #

修改device.py

### Device API Router
from fastapi import APIRouter
from routers.operations import backup
# 直接导入全局单例连接池
from db import db_pool

router = APIRouter()

# 挂载子路由
router.include_router(backup.router)

修改backup.py

from fastapi import APIRouter
# 导入同一个单例连接池
from db import db_pool

router = APIRouter(prefix="/backup")

@router.get("/status")
async def get_backup_status():
    # 使用共享连接池执行数据库操作
    async with db_pool.pool.acquire() as conn:
        async with conn.cursor() as cur:
            await cur.execute("SELECT * FROM backup_records LIMIT 10")
            records = await cur.fetchall()
    return {"backup_records": records}

这种方式下，不管多少个子路由文件导入db_pool，都是同一个连接池实例，完全避免重复创建。

方案二：使用FastAPI依赖注入（符合框架最佳实践） #

利用FastAPI的依赖注入机制管理连接池生命周期，确保单例且自动销毁。

1. 在db.py中定义连接池依赖 #

# db.py
import asyncmy
from fastapi import Depends

async def get_db_pool():
    # 初始化连接池，FastAPI默认会将该依赖作为单例处理
    pool = await asyncmy.pool.create_pool(
        host="localhost",
        user="your_db_user",
        password="your_db_pass",
        db="device_management",
        minsize=10,
        maxsize=20
    )
    yield pool
    # 应用关闭时自动关闭连接池
    await pool.close()

2. 在子路由中通过依赖使用连接池 #

修改device.py

from fastapi import APIRouter
from routers.operations import backup

router = APIRouter()
router.include_router(backup.router)

修改backup.py

from fastapi import APIRouter, Depends
from db import get_db_pool

router = APIRouter(prefix="/backup")

@router.get("/status")
async def get_backup_status(pool = Depends(get_db_pool)):
    async with pool.acquire() as conn:
        async with conn.cursor() as cur:
            await cur.execute("SELECT * FROM backup_records LIMIT 10")
            records = await cur.fetchall()
    return {"backup_records": records}

这种方式由FastAPI统一管理连接池的创建与销毁，更贴合框架设计理念，同时保证单例特性。

内容的提问来源于stack exchange，提问作者Silviu Bajenaru Marcu