FastAPI迁移至asyncpg：对应SQLAlchemy Session的会话对象及缓存排除

要排除asyncpg的数据库会话对象，你需要指定asyncpg.Connection类型，而非连接池（ConnectionPool）。因为在FastAPI中使用asyncpg时，传入接口函数的通常是单个连接实例，而非连接池本身。

具体操作 #

1. 导入asyncpg.Connection类型
2. 在fastapi_redis_cache的缓存装饰器中，将asyncpg.Connection添加到ignore_arg_types参数列表

示例代码 #

import asyncpg
from fastapi import FastAPI, Depends
from fastapi_redis_cache import FastApiRedisCache, cache

app = FastAPI()

# 初始化Redis缓存
@app.on_event("startup")
async def startup():
    redis_cache = FastApiRedisCache()
    await redis_cache.init(
        host_url="redis://localhost:6379",
        prefix="myapp-cache"
    )

# 获取asyncpg连接的依赖
async def get_db_connection() -> asyncpg.Connection:
    conn = await asyncpg.connect(user="user", password="pass", database="db", host="localhost")
    try:
        yield conn
    finally:
        await conn.close()

# 使用缓存，排除asyncpg.Connection类型
@app.get("/data/{item_id}")
@cache(ignore_arg_types=[asyncpg.Connection])
async def get_data(item_id: int, db: asyncpg.Connection = Depends(get_db_connection)):
    # 数据库查询逻辑
    result = await db.fetchrow("SELECT * FROM items WHERE id = $1", item_id)
    return result

为什么ConnectionPool不生效？ #

asyncpg.pool.Pool是连接池对象，但实际业务逻辑中传入函数的是从连接池获取的单个Connection实例。缓存键生成只会识别传入函数的参数类型，所以排除Pool类型无效，必须排除实际传入的Connection类型。

如果你的场景确实需要传入连接池作为参数，可以同时添加asyncpg.pool.Pool到ignore_arg_types里，但绝大多数场景下只需排除asyncpg.Connection即可。

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