FastAPI Router依赖Async数据库连接时的循环导入问题求解

问题场景 #

使用asyncpg初始化ConnectionPool，通过FastAPI的lifespan生命周期管理连接池，main.py内的端点可正常使用数据库，但将端点迁移到analytics.py的APIRouter时，出现循环导入错误：

Import Error: cannot import name 'get_db_connection' from partially initialized module 'app.main' (most likely due to a circular import)

当前文件结构：

.
├── app
│   ├── __init__.py
│   ├── main.py
│   └── analytics
│       ├── __init__.py
│       └── analytics.py

解决方案 #

核心是将数据库相关的依赖逻辑抽离到独立模块，打破main与analytics之间的循环依赖。

1. 创建独立的数据库依赖模块 #

在app目录下新建db.py，将连接池相关的依赖函数移至此处：

from fastapi import Depends, Request
from asyncpg.pool import Pool

async def get_database(request: Request):
    # 通过Request对象获取应用实例的连接池
    yield request.app.state.db_pool

async def get_db_connection(pool: Pool = Depends(get_database)):
    # 从连接池获取连接并自动管理释放
    async with pool.acquire() as connection:
        yield connection

2. 调整main.py代码 #

删除原有的依赖函数，导入新的db模块内容，并确保路由导入顺序合理：

from fastapi import FastAPI
from app.analytics.analytics import router as analytics_router
import asyncpg

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 初始化连接池（替换为你的数据库连接参数）
    app.state.db_pool = await asyncpg.create_pool(
        user="your_user",
        password="your_password",
        database="your_db",
        host="localhost"
    )
    yield
    # 关闭连接池
    await app.state.db_pool.close()

app = FastAPI(lifespan=lifespan)

# 导入并注册路由
app.include_router(analytics_router)

@app.get("/")
async def hello():
    return {"message": "Hello world!"}

3. 修改analytics.py的依赖导入 #

从新的db模块导入get_db_connection，而非直接从main.py导入：

from fastapi import APIRouter, Depends
from app.db import get_db_connection

router = APIRouter(
    prefix="/analytics"
)

@router.post("/test")
async def test_analytics(input_data, db = Depends(get_db_connection)):
    return "OK"

原理说明 #

原循环导入的原因是：main.py导入analytics.router，而analytics.py又导入main.py中的get_db_connection，形成闭环。将依赖逻辑抽离到独立的db.py后，main.py和analytics.py均依赖db.py，而非互相依赖，彻底打破了循环。

内容的提问来源于stack exchange，提问作者Guillermo Moreno Castaño