FastAPI使用@alru_cache时Depends无法解析协程依赖的问题

问题根源 #

你的问题核心在于三个矛盾点：

1. Foo的构造必须在运行中的事件循环上下文执行
2. 需要全局共享同一个Foo实例（单例）
3. 直接用异步缓存装饰器会返回协程对象，同步函数又无法获取事件循环

下面提供两种可靠的解决方案：

方案一：利用FastAPI生命周期事件创建单例 #

通过FastAPI的startup事件在应用启动时初始化Foo实例，并存入应用状态中，所有端点通过依赖注入共享该实例。

代码示例 #

import asyncio
from fastapi import FastAPI, Request, Depends

class Foo:
    def __init__(self):
        # 构造方法中获取当前运行的事件循环（满足要求）
        self.loop = asyncio.get_running_loop()
        self.value = "全局唯一的Foo实例"

app = FastAPI()

# 应用启动时初始化单例
@app.on_event("startup")
async def init_foo():
    # startup事件在异步上下文执行，确保有运行的事件循环
    app.state.foo = Foo()

# 定义依赖函数，从应用状态中获取单例
def get_foo(request: Request) -> Foo:
    return request.app.state.foo

# 端点使用依赖
@app.get("/")
async def get_foo_value(foo: Foo = Depends(get_foo)):
    return {"foo_value": foo.value}

优势 #

• 符合FastAPI的生命周期管理规范，实例仅在启动时创建一次
• 无需额外处理并发问题，启动逻辑天然串行执行
• 可配合shutdown事件清理Foo实例资源

方案二：异步锁+双重检查实现单例依赖 #

通过异步锁保证Foo实例仅被初始化一次，依赖函数自身管理单例逻辑，无需修改应用启动事件。

代码示例 #

import asyncio
from fastapi import FastAPI, Depends

class Foo:
    def __init__(self):
        self.loop = asyncio.get_running_loop()
        self.value = "全局唯一的Foo实例"

app = FastAPI()

# 全局变量存储单例和锁
_foo_instance = None
_foo_init_lock = asyncio.Lock()

async def get_foo() -> Foo:
    global _foo_instance
    # 第一次检查，避免每次都加锁
    if _foo_instance is None:
        async with _foo_init_lock:
            # 双重检查，防止并发请求导致多次初始化
            if _foo_instance is None:
                _foo_instance = Foo()
    return _foo_instance

# 端点使用依赖
@app.get("/")
async def get_foo_value(foo: Foo = Depends(get_foo)):
    return {"foo_value": foo.value}

优势 #

• 依赖函数自身完成单例管理，无需侵入应用启动逻辑
• 异步锁保证并发场景下仅创建一次实例
• 灵活度高，适合需要延迟初始化的场景

为什么之前的方法失败 #

1. @alru_cache装饰异步依赖：异步缓存装饰器会缓存协程对象而非最终的Foo实例，导致注入时拿到的是未执行的协程，触发AttributeError
2. 同步依赖函数：同步函数运行在非异步上下文，无法获取当前事件循环，触发RuntimeError
3. 移除缓存：每次调用依赖函数都会创建新的Foo实例，无法实现单例效果

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