FastAPI 的缓存装饰器

Question

我有一个 FastAPI 路线。

from functools import wraps

@router.get("/query/", status_code=200,)
async def query(q: str):

    # Grab info from database using query string.

    return res

上面是简化的，但准确地表达了我通常想要在我的 FastAPI 项目中对各种函数执行的操作。我创建了以下 Python 装饰器，我相信这应该是这样，但我不确定。

def cache(func):

    @wraps(func)
    def wrapper(*args, **kwargs):
        # Cache URL

    return wrapper

我正在使用 Redis 实例来缓存 FastAPI 路由中的 URL。不过，我不知道如何从

@router

装饰器获取 URL 并使用函数中的

q

参数。例如，我如何获取

"/query/"

部分和

q: str

的值并使用它我的装饰器。可能有一种方法可以将 URL 存储在变量中，然后将其传递给每个装饰器，但我正在寻找执行操作的预期方法。换句话说，我怎样才能最简洁地做到这一点而不需要太多改变我的所有功能？

Answer 1

函数的

__name__

属性（及其参数）应该用于任何缓存目的。

首先，您需要配置 Redis。

store = redis.Redis(
    host=REDIS_SERVER_URL,
    port=REDIS_PORT,
    decode_responses=True,
)

然后，这是一个应该可以工作的缓存功能。您还可以传入缓存的小时数（默认为两小时）。

def cache(_, hours=2):
    def wrapper(func):
        @wraps(func)
        def wrapped(*args, **kwargs):
            key_parts = [func.__name__] + list(args)
            key = "-".join(str(k) for k in key_parts)
            result = store.get(key)
            if result is None:
                value = func(*args, **kwargs)
                value_json = json.dumps(value)
                expire_time = 60 * 60 * hours
                store.setex(key, expire_time, value_json)
            else:
                value_json = result.decode("utf-8")
                value = json.loads(value_json)
            return value
        return wrapped
    return wrapper

@cache # Cache for two hours.
@router.get("/", status_code=200,)
async def root():
    return {"message": "Hello World!"}

对于我的项目，我还需要缓存协程异步函数，所以这就是。

def cache(_, hours=2):
    def wrapper(func):
        @wraps(func)
        async def wrapped(*args, **kwargs):
            key_parts = [func.__name__] + list(args)
            key = "-".join(str(k) for k in key_parts)
            result = store.get(key)
            if result is None:
                is_coroutine = iscoroutinefunction(func)
                if is_coroutine:
                    value = await func(*args, **kwargs)
                else:
                    value = func(*args, **kwargs)
                value_json = json.dumps(value)
                expire_time = 60 * 60 * hours
                store.setex(key, expire_time, value_json)
            else:
                value_json = result.decode("utf-8")
                value = json.loads(value_json)
            return value
        return wrapped
    return wrapper