一、FastAPI Cache 配置

要在 FastAPI 中使用缓存，我们可以使用 fastapi-cache 这个第三方库。下面是配置 fastapi-cache 的步骤：

1. 安装 fastapi-cache

首先，安装 fastapi-cache 库。

1
pip install fastapi-cache

2. 配置缓存后端

fastapi-cache 支持多种缓存后端，如 Redis、Memcached 等。以下是一个使用 Redis 作为缓存后端的配置示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
from fastapi import FastAPI
from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from fastapi_cache.decorator import cache

app = FastAPI()

# 配置 Redis 缓存后端
FastAPICache.init(
    backend=RedisBackend(redis_url="redis://localhost:6379/1"),
    prefix="fastapi-cache",
    timeout=60
)

在上面的配置中，我们指定了 Redis 的连接地址，并设置了缓存前缀和超时时间。

二、使用 FastAPI Cache

配置好缓存后，我们可以使用 @cache 装饰器来装饰 FastAPI 路径操作，使其结果被缓存。

1. 缓存路径操作

以下是一个简单的示例，展示如何缓存一个路径操作的结果：

1
2
3
4
5
@app.get("/items/{item_id}")
@cache()
async def read_item(item_id: int):
    # 模拟数据库查询
    return {"item_id": item_id, "name": "Item Name"}

在这个例子中，read_item 函数的结果将被缓存。当同一个 item_id 的请求再次到来时，FastAPI 将直接返回缓存的结果，而不会执行函数。

2. 缓存参数

@cache 装饰器支持多个参数，用于自定义缓存行为：

  • timeout：缓存超时时间，默认为配置中的 timeout。
  • key：自定义缓存键，默认为请求路径和查询参数。
  • unless：一个函数，如果返回 True，则不缓存结果。

以下是一个使用自定义缓存键的示例：

1
2
3
4
5
@app.get("/items/{item_id}")
@cache(key="item_{item_id}", timeout=120)
async def read_item(item_id: int):
    # 模拟数据库查询
    return {"item_id": item_id, "name": "Item Name"}

三、使用异步redis库

我们还可以使用redis的asyncio版本，并建立一个redis 连接池。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
import redis.asyncio as redis
from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend

@asynccontextmanager
async def build_redis_cache():
    """配置Redis缓存, 并初始化FastAPICache."""
    # 创建redis 连接池链接
    pool = redis.ConnectionPool.from_url(
        "redis://localhost:6379/1",
        encoding="utf8",
        decode_responses=True,
    )
    client = redis.Redis(connection_pool=pool)
    # 初始化FastAPICache
    FastAPICache.init(RedisBackend(client), prefix="fastapi-cache")

    yield client

    await client.aclose()
    await pool.aclose()