long2ice / fastapi-cache

fastapi-cache is a tool to cache fastapi response and function result, with backends support redis and memcached.
https://github.com/long2ice/fastapi-cache
Apache License 2.0
1.34k stars 162 forks source link

Feature Request: use Annotation markers to special-case argument handling #129

Open mjpieters opened 1 year ago

mjpieters commented 1 year ago

The context

Currently, if your cached endpoint uses arguments that should be ignored when generating a cache key, you have no choice but to create a custom key builder.

E.g. the following endpoint will almost never return a cached response, because the background_tasks dependency will rarely produce the same cache key value (example based on the FastAPI documentation):

from contextlib import asynccontextmanager
from fastapi import BackgroundTasks, FastAPI
from fastapi_cache.decorator import cache

@asynccontextmanager
async def lifespan():
    redis = aioredis.from_url("redis://localhost", encoding="utf8", decode_responses=True)
    FastAPICache.init(RedisBackend(redis), prefix="fastapi-cache")
    yield
    FastAPICache.reset()

app = FastAPI(lifespan=lifespan)

def write_notification(email: str, message=""):
    with open("log.txt", mode="w") as email_file:
        content = f"notification for {email}: {message}"
        email_file.write(content)

@app.post("/send-notification/{email}")
@cache(expire=60)
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, message="some notification")
    return {"message": "Notification sent in the background"}

You'd have to write a custom key builder here; you could still delegate to the default keybuilder after removing certain arguments:

from typing import Any, Callable

from fastapi import BackgroundTasks
from fastapi.dependencies.utils import get_typed_signature
from fastapi_cache import default_key_builder

def custom_key_builder(
    func: Callable[..., Any], namespace: str, *, kwargs: dict[str, Any], **kw: Any
) -> str:
    # ignore the task argument

    for param in get_typed_signature(func).parameters.items():
        if param.annotation is BackgroundTasks:
            kwargs.pop(param.name, None)
    return default_key_builder(func, namespace, kwargs=kwargs, **kw)

The above key builder is generic enough that it can be used as the key builder for the whole project at least.

Annotated arguments

But, what if you could simply annotate arguments that should not be part of the key?

It could look something like this:

from typing import Annotated
from fastapi_cache import IgnoredArg

@app.post("/send-notification/{email}")
@cache(expire=60)
async def send_notification(email: str, background_tasks: Annotated[BackgroundTasks, IgnoredArg]):
    background_tasks.add_task(write_notification, email, message="some notification")
    return {"message": "Notification sent in the background"}

IgnoredArg is just a sentinel object here:

IgnoredArg = object()

The cache decorator could trivially filter out such arguments by introspecting the endpoint signature:

from typing import Annotated, get_args, get_origin
from fastapi.dependencies.utils import get_typed_signature

ignored = set()
for param in get_typed_signature(func).items():
    ann = param.annotation
    if get_origin(ann) is Annotated and IgnoredArg in get_args(ann):
        ignored.append(param.name)

and the ignored set can then be used to remove names from the kwargs dictionary before passing it to the key builder.

Annotations to convert argument values to key components

You could take this concept another step further, and use such annotations to register argument converters to produce key components. Another example:

from fastapi_cache import AsKey, IgnoredArg

def canonical_email(email: str) -> str:
     username, _, domain = email.rpartition('@')[-1]
     username, domain = username.strip(), domain.strip().lower()
     if domain == "google.com":
         username = username.replace(".", "").partition("+")[0]
    return f"{username}@{domain}"

@app.post("/send-notification/{email}")
@cache(expire=60)
async def send_notification(
    email: Annotated[str, AsKey(canonical_email)],
    background_tasks: Annotated[BackgroundTasks, IgnoredArg]
):
    background_tasks.add_task(write_notification, email, message="some notification")
    return {"message": "Notification sent in the background"}

The cache decorator would then pass the email argument through the canonical_email() callable before passing on the arguments to the key builder.

mjpieters commented 1 year ago

One caveat: currently you can't use Annotated on BackgroundTasks or any of the other non-field parameters. I've requested that FastAPI fix this.

mjpieters commented 1 year ago

I can't, currently, see any use case where a non-field parameter should be part of the cache key, so I'm going to treat those as IgnoreArg by default and document this.