如何利用FastAPI的lifespan管理Redis连接池并保留依赖注入便捷性？

当前我使用FastAPI的依赖注入来获取Redis连接，但连接池是在Lifespan外部初始化的。希望改用Lifespan来管理连接池的创建与销毁，同时不想放弃Depends(get_redis)的便捷性，避免在接口里直接调用request.app.state。

当前实现如下：

接口依赖定义 #

@router.post("/items")
async def post_items(r: RedisDependency):
    ...

RedisDependency定义 #

RedisDependency = Annotated[Redis, Depends(get_redis)]

get_redis函数实现 #

import redis.asyncio as aredis

from . import pool


async def get_redis() -> aredis.Redis:
    r = aredis.Redis(connection_pool=pool, decode_responses=True)
    try:
        yield r
    finally:
        await r.aclose()

连接池初始化（db/init.py） #

import redis.asyncio as aredis

from app.core.config import settings

def create_redis_pool():
    return aredis.ConnectionPool.from_url(settings.REDIS_URL)

pool = create_redis_pool()

可以通过以下方式实现：在Lifespan中创建连接池并绑定到app.state，然后修改依赖函数从app.state获取连接池，同时保持接口的依赖注入方式不变。

1. 实现Lifespan管理连接池 #

在FastAPI应用实例化时，添加Lifespan上下文管理器，负责连接池的创建与销毁：

from fastapi import FastAPI
import redis.asyncio as aredis
from app.core.config import settings

async def lifespan(app: FastAPI):
    # 启动时创建连接池
    app.state.redis_pool = aredis.ConnectionPool.from_url(settings.REDIS_URL)
    yield
    # 关闭时销毁连接池
    await app.state.redis_pool.disconnect()

app = FastAPI(lifespan=lifespan)

2. 修改get_redis依赖函数 #

让依赖函数通过Request对象获取app.state里的连接池，这样接口层无需改动：

from fastapi import Depends, Request
import redis.asyncio as aredis
from typing import Annotated

async def get_redis(request: Request) -> aredis.Redis:
    r = aredis.Redis(
        connection_pool=request.app.state.redis_pool,
        decode_responses=True
    )
    try:
        yield r
    finally:
        await r.aclose()

RedisDependency = Annotated[aredis.Redis, Depends(get_redis)]

3. 接口保持原有写法 #

无需修改接口代码，依然通过依赖注入获取Redis实例：

@router.post("/items")
async def post_items(r: RedisDependency):
    # 正常使用r操作Redis
    await r.set("key", "value")
    ...

4. 移除原有db模块的全局连接池 #

删除db/__init__.py里的全局pool变量，避免冗余初始化。

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