FastAPI多路由共享单例数据库连接：如何避免循环依赖？

核心需求是启动时仅初始化一次数据库连接，同时解决main.py与路由模块的循环依赖问题，以下是几种符合Python风格的规范实现方案：

方案1：独立数据库模块（最推荐） #

将数据库初始化逻辑抽离到独立模块，让main.py和路由模块都从该模块导入实例，彻底打破循环依赖，这是Python模块化设计的标准做法。

实现步骤： #

1. 创建app/db.py模块，专门管理数据库连接：

# app/db.py
# 示例用字符串模拟数据库实例，实际场景替换为真实连接逻辑
some_db_instance = "Initiate DB Instance once on App startup."

# 生产场景示例（SQLAlchemy同步连接）：
# from sqlalchemy import create_engine
# from sqlalchemy.orm import sessionmaker
#
# # 启动时仅创建一次数据库引擎
# engine = create_engine("postgresql://user:password@localhost/dbname")
# # 会话工厂，每次请求创建新会话但复用引擎
# SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

2. 修改app/main.py，移除内部数据库实例定义，按需从db.py导入：

# app/main.py
import fastapi
from fastapi import FastAPI
from app.routers.test_router import router
# 若main中需要使用数据库实例，直接从db模块导入
from app.db import some_db_instance

app = FastAPI()
app.include_router(router)

@app.get("/", include_in_schema=False)
async def root():
    return fastapi.responses.RedirectResponse("/docs")

3. 修改app/routers/test_router.py，从db.py导入数据库实例：

# app/routers/test_router.py
from fastapi import APIRouter
from app.db import some_db_instance

router = APIRouter(prefix="/test", tags=["Test"])

@router.get("/data")
async def get_test():
    do_stuff_with_db = some_db_instance + "..."
    return {"result": do_stuff_with_db}

优势： #

• 代码结构清晰，职责分离，符合单一原则
• 彻底消除循环依赖，模块间依赖关系明确
• 数据库实例仅在模块加载时初始化一次，满足单例需求

方案2：使用FastAPI依赖注入（适合异步/复杂场景） #

如果需要异步初始化数据库、管理连接池，或者希望更灵活地控制实例生命周期，FastAPI的依赖注入系统是最佳选择。

实现步骤： #

1. 创建app/db.py，封装数据库初始化和获取逻辑：

# app/db.py
from typing import AsyncGenerator

# 存储全局数据库实例
_db_instance = None

async def init_db():
    # 实际异步数据库初始化逻辑（示例用字符串模拟）
    # import asyncpg
    # conn = await asyncpg.connect(user="user", password="pwd", database="db", host="127.0.0.1")
    conn = "Async DB Instance initialized once"
    return conn

# 依赖函数，确保实例仅初始化一次
async def get_db() -> AsyncGenerator[str, None]:
    global _db_instance
    if not _db_instance:
        _db_instance = await init_db()
    yield _db_instance

2. 修改app/main.py，在启动事件中预初始化数据库（可选，也可让第一个请求触发初始化）：

# app/main.py
import fastapi
from fastapi import FastAPI
from app.routers.test_router import router
from app.db import init_db

app = FastAPI()
app.include_router(router)

# 启动时触发初始化，确保服务就绪时数据库已连接
@app.on_event("startup")
async def startup_event():
    await init_db()

@app.get("/", include_in_schema=False)
async def root():
    return fastapi.responses.RedirectResponse("/docs")

3. 修改app/routers/test_router.py，通过Depends获取数据库实例：

# app/routers/test_router.py
from fastapi import APIRouter, Depends
from app.db import get_db

router = APIRouter(prefix="/test", tags=["Test"])

@router.get("/data")
async def get_test(db=Depends(get_db)):
    do_stuff_with_db = db + "..."
    return {"result": do_stuff_with_db}

优势： #

• 适配异步数据库场景，符合FastAPI的异步设计理念
• 依赖注入解耦了路由与数据库实例，便于测试（可替换测试用实例）
• 自动管理实例生命周期，确保仅初始化一次

方案3：延迟导入（临时过渡，不推荐） #

如果不想调整现有代码结构，可临时采用函数内部延迟导入的方式规避循环依赖，但该方案会降低代码可读性，不适合长期维护。

修改app/routers/test_router.py：

# app/routers/test_router.py
from fastapi import APIRouter

router = APIRouter(prefix="/test", tags=["Test"])

@router.get("/data")
async def get_test():
    # 仅在函数调用时导入，避免模块加载阶段的循环依赖
    from app.main import some_db_instance
    do_stuff_with_db = some_db_instance + "..."
    return {"result": do_stuff_with_db}

劣势： #

• 模块依赖关系隐藏在函数内部，不利于代码维护和调试
• 每次请求都会执行导入操作（Python会缓存模块，实际性能影响小，但逻辑不严谨）

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