当前位置：首页 > 文章列表 > 文章 > python教程 > FastAPI启动事件依赖注入技巧

FastAPI启动事件依赖注入技巧

2025-10-27 09:06:32 0浏览收藏

本文深入探讨了在FastAPI应用中使用启动事件进行依赖注入的实践，着重指出了直接在`@app.on_event("startup")`中使用`Depends()`与`AsyncGenerator`初始化资源（如Redis连接）时可能遇到的问题，并明确了`Depends()`在此场景下的局限性。文章核心在于详细阐述了如何利用FastAPI的`lifespan`上下文管理器，以更优雅、更可靠的方式管理异步生成器依赖，确保应用启动时资源得到正确初始化，有效避免`AttributeError`等错误。通过对比错误示例和正确实践，强调了`lifespan`在应用生命周期管理中的重要作用，为开发者提供了在FastAPI项目中进行资源初始化和依赖管理的有效解决方案，助力构建更健壮、更高效的异步应用。

FastAPI启动事件中AsyncGenerator依赖注入的正确实践

本文探讨了在FastAPI应用的startup事件中直接使用Depends()与AsyncGenerator进行资源（如Redis连接）初始化时遇到的问题，并指出Depends()不适用于此场景。核心内容是提供并详细解释了如何通过FastAPI的lifespan上下文管理器来正确、优雅地管理异步生成器依赖，确保应用启动时资源正确初始化，避免AttributeError。

问题背景：在应用启动时初始化异步资源

在构建基于FastAPI的异步应用时，我们经常需要在应用启动时初始化一些全局资源，例如数据库连接池、消息队列客户端或缓存连接。FastAPI提供了@app.on_event("startup")装饰器来处理这些启动任务。同时，为了更好地管理资源生命周期，我们通常会使用异步生成器（AsyncGenerator）来创建和关闭这些资源，并结合FastAPI的依赖注入系统Depends()。然而，当尝试在startup事件中直接将AsyncGenerator与Depends()结合使用时，可能会遇到意料之外的错误。

考虑一个场景，我们需要在FastAPI应用启动时获取一个Redis异步客户端，并将其用于初始化一个全局的任务队列。我们可能尝试编写如下代码：

import uvicorn
from fastapi import FastAPI, Depends
import redis.asyncio as redis
from redis.asyncio import Redis
from typing import AsyncGenerator
from rq import Queue # 假设rq是任务队列库

# 配置Redis连接
REDIS_HOST = "localhost"
REDIS_PORT = 6379

redis_pool = redis.ConnectionPool.from_url(f"redis://{REDIS_HOST}:{REDIS_PORT}")

async def get_async_redis_client() -> AsyncGenerator[Redis, None]:
    """
    异步生成器，用于提供Redis客户端连接。
    """
    async with Redis.from_pool(redis_pool) as client:
        yield client

def process_data(data: str):
    """
    模拟一个处理数据的函数。
    """
    print(f"Processing data: {data}")

def create_app():
    app = FastAPI(docs_url='/')
    task_queue: Queue = None # 声明为None，稍后初始化

    @app.on_event("startup")
    async def startup_event(redis_conn: redis.asyncio.Redis = Depends(get_async_redis_client)):
        """
        尝试在startup事件中使用Depends()注入Redis连接。
        """
        nonlocal task_queue
        task_queue = Queue("task_queue", connection=redis_conn)
        print("Redis connection initialized in startup event.")

    @app.post("/add_data")
    async def add_data(data: str):
        """
        添加数据到任务队列。
        """
        if task_queue:
            task_queue.enqueue(process_data, data)
            return {"message": "Book in processing"}
        return {"message": "Task queue not initialized", "status": "error"}

    @app.get("/get_data")
    async def get_data():
        """
        示例接口。
        """
        return {"data": "kek"}

    return app

def main():
    uvicorn.run(
        f"{__name__}:create_app",
        host='0.0.0.0', port=8888,
        reload=True
    )

if __name__ == '__main__':
    main()

当运行上述代码并尝试向/add_data端点发送POST请求时，会收到一个AttributeError: 'Depends' object has no attribute 'pipeline'的错误。这表明在startup_event函数中，redis_conn变量并没有被解析成实际的redis.asyncio.Redis对象，而仍然是一个Depends对象。

理解FastAPI的依赖注入与启动事件

FastAPI的Depends()机制主要设计用于请求处理函数中的依赖解析。在请求处理的生命周期中，FastAPI会负责调用依赖函数（包括异步生成器），获取其yield出的值，并在请求结束后执行生成器中yield之后的清理代码。

然而，@app.on_event("startup")装饰器下的函数，其执行时机在整个应用开始接受请求之前，并且它不属于标准的请求-响应循环。FastAPI的依赖注入系统并不会像处理路由函数那样，自动解析startup事件函数参数中的Depends()。因此，redis_conn变量接收到的不是get_async_redis_client生成器yield出的Redis客户端实例，而是Depends(get_async_redis_client)这个Depends对象本身。当rq库尝试对这个Depends对象调用pipeline()方法时，自然会抛出AttributeError。

正确实践：利用lifespan管理异步生成器依赖

为了在应用启动时正确地初始化和管理异步生成器提供的资源，FastAPI推荐使用lifespan上下文管理器。lifespan是FastAPI 0.65.0版本引入的一种更现代、更灵活的应用生命周期管理方式，它允许我们定义一个异步上下文管理器，在应用启动前执行设置代码，并在应用关闭前执行清理代码。

通过lifespan，我们可以手动调用异步生成器，获取其yield出的资源，并将其存储在应用实例或全局变量中，供其他部分使用。

以下是使用lifespan解决上述问题的正确方法：

import uvicorn
from fastapi import FastAPI
import redis.asyncio as redis
from redis.asyncio import Redis
from typing import AsyncGenerator
from rq import Queue # 假设rq是任务队列库
from contextlib import asynccontextmanager

# 配置Redis连接
REDIS_HOST = "localhost"
REDIS_PORT = 6379

redis_pool = redis.ConnectionPool.from_url(f"redis://{REDIS_HOST}:{REDIS_PORT}")

async def get_async_redis_client() -> AsyncGenerator[Redis, None]:
    """
    异步生成器，用于提供Redis客户端连接。
    """
    print("Opening Redis connection...")
    async with Redis.from_pool(redis_pool) as client:
        yield client
    print("Closing Redis connection...") # 应用关闭时执行

def process_data(data: str):
    """
    模拟一个处理数据的函数。
    """
    print(f"Processing data: {data}")

# 定义一个全局变量来存储任务队列
task_queue: Queue = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    """
    FastAPI应用生命周期管理器。
    在应用启动时初始化资源，在应用关闭时清理资源。
    """
    global task_queue # 声明使用全局变量

    # 手动调用异步生成器以获取Redis连接
    # 注意：这里直接调用get_async_redis_client()，并迭代它
    # app.dependency_overrides.get(get_async_redis_client, get_async_redis_client)
    # 这一步是为了兼容可能存在的依赖覆盖，确保获取到的是最终的依赖函数
    redis_generator_func = app.dependency_overrides.get(get_async_redis_client, get_async_redis_client)
    async for redis_conn in redis_generator_func():
        # 在这里，redis_conn已经是实际的Redis客户端对象
        task_queue = Queue("task_queue", connection=redis_conn)
        print("Redis connection and Task Queue initialized via lifespan.")
        yield # 应用在此处启动并处理请求
    # 应用关闭时，生成器会继续执行，清理Redis连接
    print("Application shutdown: Resources released.")

def create_app():
    app = FastAPI(
        docs_url='/',
        lifespan=lifespan # 将lifespan上下文管理器传递给FastAPI
    )

    @app.post("/add_data")
    async def add_data(data: str):
        """
        添加数据到任务队列。
        """
        if task_queue:
            task_queue.enqueue(process_data, data)
            return {"message": "Book in processing"}
        return {"message": "Task queue not initialized", "status": "error"}

    @app.get("/get_data")
    async def get_data():
        """
        示例接口。
        """
        return {"data": "kek"}

    return app

def main():
    uvicorn.run(
        f"{__name__}:create_app",
        host='0.0.0.0', port=8888,
        reload=True
    )

if __name__ == '__main__':
    main()

在这个修正后的代码中：

@asynccontextmanager装饰器: 我们使用contextlib.asynccontextmanager装饰器将lifespan函数转换为一个异步上下文管理器。
lifespan函数: 这个函数现在负责整个应用的生命周期。
- 在yield之前，它执行应用启动时的初始化逻辑。我们在这里手动调用get_async_redis_client()异步生成器，并通过async for循环获取yield出的redis_conn对象。
- app.dependency_overrides.get(get_async_redis_client, get_async_redis_client)这一步是为了确保即使get_async_redis_client被覆盖（例如在测试环境中），lifespan也能获取到正确的依赖函数。
- task_queue被正确地用解析后的redis_conn对象初始化。
- yield语句将控制权交还给FastAPI，此时应用开始处理请求。
- 当应用关闭时（例如通过发送中断信号），yield之后的代码会被执行，从而触发get_async_redis_client生成器中的清理逻辑（async with Redis.from_pool(redis_pool) as client:块的退出）。
FastAPI(lifespan=lifespan): 在创建FastAPI应用实例时，通过lifespan参数注册我们定义的生命周期管理器。

注意事项

全局变量管理: 在lifespan函数中修改全局变量（如task_queue）时，务必使用global关键字来指示您正在修改全局作用域的变量，而不是创建局部变量。
依赖覆盖兼容性: app.dependency_overrides.get(get_async_redis_client, get_async_redis_client)是一个健壮的做法。它首先检查get_async_redis_client是否被app.dependency_overrides覆盖。如果没有，它就使用原始的get_async_redis_client函数。这对于测试和更复杂的应用场景非常有用。
资源清理: 使用AsyncGenerator结合async with语句是管理异步资源生命周期的推荐方式。lifespan上下文管理器确保了AsyncGenerator的清理部分在应用关闭时被正确执行。
startup事件与lifespan: 尽管@app.on_event("startup")仍然可用，但对于需要复杂初始化和清理逻辑的资源，或者需要与依赖注入系统交互的场景，lifespan提供了更强大、更清晰的机制。

总结

在FastAPI中，Depends()装饰器是为请求处理函数设计的依赖注入机制，不适用于@app.on_event("startup")事件。当需要在应用启动时利用AsyncGenerator初始化全局资源时，正确的做法是使用FastAPI的lifespan上下文管理器。通过手动调用异步生成器并将其结果存储在全局变量中，我们可以确保资源在应用启动时被正确初始化，并在应用关闭时被优雅地清理，从而避免因依赖解析不当导致的AttributeError。这种方法不仅解决了特定问题，也体现了FastAPI在应用生命周期管理上的灵活性和强大功能。

今天关于《FastAPI启动事件依赖注入技巧》的内容介绍就到此结束，如果有什么疑问或者建议，可以在golang学习网公众号下多多回复交流；文中若有不正之处，也希望回复留言以告知！