本文详细介绍了如何在fastapi应用中，利用其`lifespan`生命周期管理功能，结合依赖注入（`depends`），优雅地创建和释放redis连接池。通过将连接池的初始化与清理绑定到应用启动和关闭事件，并设计一个从该池获取redis客户端的依赖函数，实现了资源的高效复用与统一管理，同时避免了在路由中直接访问`request.app.state`的复杂性，提升了代码的可维护性与专业性。
引言：FastAPI应用中的资源管理挑战
在构建高性能的异步Web服务时，高效地管理外部资源（如数据库连接池、缓存客户端、消息队列连接等）至关重要。这些资源通常需要在应用程序启动时进行初始化，并在应用程序关闭时进行清理，以确保资源的正确分配和释放，避免资源泄露。FastAPI作为一款现代、高性能的Web框架，提供了强大的依赖注入系统，但如何将全局资源的生命周期管理与依赖注入机制完美结合，是开发者需要面对的一个常见问题。
本文将聚焦于如何在FastAPI中，利用其lifespan功能来管理Redis连接池的创建与释放，并结合依赖注入，使得在路由函数中能够以简洁、一致的方式获取Redis客户端实例，同时避免直接在业务逻辑中暴露底层资源管理细节。
FastAPI的生命周期管理：lifespan
FastAPI提供了lifespan（或旧版中的on_event("startup")和on_event("shutdown")）机制，允许开发者在应用程序启动和关闭时执行自定义逻辑。这是一个理想的场所来初始化和清理应用程序所需的全局资源。
lifespan是一个异步上下文管理器，它在FastAPI应用程序启动时进入，在应用程序关闭时退出。其基本结构如下：
from contextlib import asynccontextmanager
from fastapi import FastAPI
@asynccontextmanager
async def lifespan(app: FastAPI):
# 应用程序启动时执行的逻辑，例如初始化数据库连接池
print("Application startup: Initializing resources...")
app.state.my_resource = "Initialized Resource" # 将资源存储在 app.state
yield
# 应用程序关闭时执行的逻辑，例如关闭数据库连接池
print("Application shutdown: Cleaning up resources...")
# 清理 app.state.my_resource通过将资源存储在app.state对象上，我们可以在应用程序的任何地方（包括依赖项）访问这些全局资源。
使用lifespan管理Redis连接池
我们将以redis.asyncio库为例，演示如何使用lifespan来管理Redis连接池，并结合FastAPI的依赖注入系统。
1. 配置Redis连接URL
首先，我们需要一个地方来存储Redis的连接URL。推荐使用Pydantic Settings进行配置管理。
app/core/config.py:
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
REDIS_URL: str = "redis://localhost:6379/0" # 默认Redis连接URL
settings = Settings()2. 定义lifespan上下文管理器
在FastAPI应用的主文件中，定义lifespan上下文管理器来初始化和关闭Redis连接池。
app/main.py:
from contextlib import asynccontextmanager
from typing import AsyncGenerator, Annotated
import redis.asyncio as aredis
from fastapi import FastAPI, Depends, APIRouter, Request
from app.core.config import settings # 导入配置
# 定义 lifespan 上下文管理器
@asynccontextmanager
async def lifespan(app: FastAPI):
"""
FastAPI 应用程序生命周期管理器。
在应用程序启动时初始化 Redis 连接池，并在关闭时将其断开。
"""
print("Initializing Redis Connection Pool...")
# 创建 Redis 连接池并存储在 app.state 中
# decode_responses=True 使得从 Redis 获取的数据自动解码为字符串
app.state.redis_pool = aredis.ConnectionPool.from_url(
settings.REDIS_URL, decode_responses=True
)
print("Redis Connection Pool Initialized.")
yield # 应用程序在此处开始处理请求
print("Closing Redis Connection Pool...")
# 应用程序关闭时，断开 Redis 连接池
await app.state.redis_pool.disconnect()
print("Redis Connection Pool Closed.")
# 创建 FastAPI 应用实例，并传入 lifespan
app = FastAPI(lifespan=lifespan)
router = APIRouter()在此步骤中，我们：
- 在lifespan函数的yield之前，使用aredis.ConnectionPool.from_url创建了一个全局的Redis连接池。
- 将这个连接池实例存储在app.state.redis_pool中，使其在整个应用生命周期内可访问。
- 在yield之后，即应用程序关闭时，调用app.state.redis_pool.disconnect()来优雅地关闭连接池，释放所有连接。
3. 创建依赖注入函数
为了在路由函数中方便地使用Redis客户端，我们需要创建一个依赖注入函数。这个函数将从app.state中获取已初始化的连接池，并为每个请求提供一个Redis客户端实例。
# app/main.py (接着上面的代码)
async def get_redis_client(request: Request) -> AsyncGenerator[aredis.Redis, None]:
"""
依赖注入函数，为每个请求提供一个 Redis 客户端实例。
该客户端从应用程序全局连接池中获取连接。
"""
# 从 app.state 中获取已初始化的 Redis 连接池
pool = request.app.state.redis_pool
# 从连接池创建一个 Redis 客户端实例
# 注意：aredis.Redis 客户端是异步安全的，并且内部会管理从连接池获取和释放连接
r = aredis.Redis(connection_pool=pool, decode_responses=True)
try:
yield r # 将 Redis 客户端提供给路由函数
finally:
# 在请求处理完毕后，关闭当前客户端实例，
# 这会将底层连接释放回连接池。
await r.aclose()
# 定义一个类型别名，方便在路由函数中使用
RedisDependency = Annotated[aredis.Redis, Depends(get_redis_client)]这个get_redis_client依赖函数是关键：
- 它接收Request对象，以便访问app.state中存储的全局连接池。
- 它利用aredis.Redis(connection_pool=pool, ...)从全局连接池创建一个新的Redis客户端实例。这个客户端实例会在其生命周期内从连接池获取和释放连接。
- 使用try...finally块确保在请求处理完毕后，调用r.aclose()来关闭当前的Redis客户端实例，这会将它所使用的连接返回到连接池中，实现连接的复用。
4. 在FastAPI应用中使用依赖
现在，我们可以在FastAPI的路由函数中，通过依赖注入的方式，简洁地获取和使用Redis客户端了。
# app/main.py (接着上面的代码)
@router.post("/items/")
async def create_item(item_data: dict, r: RedisDependency):
"""
示例路由：创建一个新的 item 并将其存储到 Redis。
"""
item_id = item_data.get("id") or "new_item"
key = f"item:{item_id}"
await r.set(key, str(item_data))
value = await r.get(key)
return {"message": "Item created and stored in Redis", "item_id": item_id, "stored_value": value}
@router.get("/items/{item_id}")
async def read_item(item_id: str, r: RedisDependency):
"""
示例路由：从 Redis 中读取一个 item。
"""
key = f"item:{item_id}"
value = await r.get(key)
if value:
return {"item_id": item_id, "data": value}
return {"message": "Item not found", "item_id": item_id}
app.include_router(router)通过r: RedisDependency，路由函数可以直接获得一个可用的Redis客户端，无需关心其初始化和清理过程。
优点与最佳实践
- 统一的资源管理：所有全局资源的生命周期（创建、初始化、清理）都由FastAPI的lifespan统一控制，确保了资源的正确管理。
- 依赖注入的便利性：通过Depends机制，路由函数无需关心资源获取的复杂细节，只需声明所需资源类型，FastAPI会自动注入，提高了代码的可读性和可维护性。
- 避免全局变量污染：将资源实例存储在app.state中，避免了直接使用全局变量，使得资源管理更加清晰和可控。
- 优雅的资源清理：lifespan确保在应用程序关闭时，连接池能够被正确断开，防止资源泄露。
- 解耦与测试性：这种模式将资源管理逻辑与业务逻辑解耦，使得单元测试和集成测试更加容易。你可以轻松地为get_redis_client依赖提供模拟实现。
- 避免直接在路由中使用request.app.state：虽然request.app.state可以访问全局资源，但将其封装在依赖函数中，使得路由函数更加专注于业务逻辑，避免了在每个路由中重复编写资源获取代码。
总结
通过结合FastAPI
