如何在FastAPI的Lifespan上下文/启动事件中注入依赖？

FastAPI的lifespan上下文管理器确实不支持直接通过Depends注入依赖，因为它不属于路由处理流程，框架不会自动解析其参数依赖。不过可以通过手动调用FastAPI的依赖解析工具，复用已有的依赖注入逻辑，无需手动创建实例。

方案一：使用solve_dependencies手动解析依赖 #

利用FastAPI内置的solve_dependencies工具，模拟请求上下文来解析你的依赖链，完全复用已定义的init_my_repository等依赖函数：

from fastapi import FastAPI, Depends
from contextlib import asynccontextmanager
from fastapi.dependencies.utils import solve_dependencies
from starlette.requests import Request

# 你的Repository和依赖定义保持不变
class Repository:
    def __init__(self, db: Database):
        self.collection = db['my_collection']

def init_my_repository(db=Depends(get_mongo_database_instance)):
    return Repository(db)

# 其他依赖示例
def init_my_XXX():
    return XXX()

def init_my_YYY():
    return YYY()

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 创建模拟请求上下文，用于依赖解析
    mock_request = Request(
        scope={
            "type": "http",
            "path": "/",
            "method": "GET",
            "query_string": b"",
            "headers": {},
            "app": app,
        },
        receive=lambda: None  # 空的receive回调，仅用于满足Request实例要求
    )
    
    # 定义需要解析的依赖列表
    required_deps = [
        Depends(init_my_repository),
        Depends(init_my_XXX),
        Depends(init_my_YYY)
    ]
    
    # 解析依赖
    resolved_deps, _, _ = await solve_dependencies(
        request=mock_request,
        dependencies=required_deps,
        dependency_overrides=app.dependency_overrides
    )
    
    repo, xxx, yyy = resolved_deps
    
    # 执行启动初始化任务
    repo.do_something()
    xxx.do_something()
    yyy.do_something()
    
    yield  # 应用运行阶段
    
    # 可选：应用关闭时的清理逻辑
    # await repo.cleanup()

app = FastAPI(lifespan=lifespan)

说明 #

• 此方法完全遵循FastAPI的依赖注入规则，包括支持dependency_overrides，和路由中的依赖行为一致
• 如果你的依赖是异步函数（比如get_mongo_database_instance是异步的），solve_dependencies会自动处理异步解析
• 模拟的Request仅用于触发依赖解析逻辑，不会实际处理请求

方案二：将依赖实例存入app.state #

如果你的依赖链不复杂，也可以手动解析依赖并将实例存入app.state，方便后续在路由或lifespan中复用：

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 手动解析顶层依赖
    db = get_mongo_database_instance()  # 如果是异步函数则加await
    app.state.repo = Repository(db)
    app.state.xxx = init_my_XXX()
    app.state.yyy = init_my_YYY()
    
    # 执行初始化
    app.state.repo.do_something()
    app.state.xxx.do_something()
    app.state.yyy.do_something()
    
    yield
    
    # 清理逻辑
    # del app.state.repo

说明 #

• 这种方式更简洁，但如果依赖层级较深（比如init_my_XXX本身还有其他依赖），需要手动逐层解析，不如方案一灵活
• 存入app.state的实例可以在路由中通过request.app.state.repo访问

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