Description
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.