[C27Cache](https://pypi.org/project/c27cache/) is a simple HTTP caching library designed to work with [FastAPI](https://fastapi.tiangolo.com/).
C27Cache is a simple HTTP caching library designed to work with FastAPI
While C27Cache is still early in it's development, it's been used in production on a couple of services.
pip install c27cache
poetry add c27cache
from c27cache.config import C27Cache
from pytz import timezone
asia_kolkata = timezone('Asia/Kolkata')
C27Cache.init(redis_url=REDIS_URL, namespace='test_namespace', tz=asia_kolkata)
The tz
attribute becomes import when the cache
decorator relies on the expire_end_of_day
and expire_end_of_week
attributes to expire the cache key.
The ttl_in_seconds
expires the cache in 180 seconds. There are other approaches to take with helpers like expire_end_of_day
and expires_end_of_week
import redis
from datetime import datetime
from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse
from c27cache import C27Cache, cache, invalidate_cache
@app.get("/b/home")
@cache(key="b.home", ttl_in_seconds=180)
async def home(request: Request, response: Response):
return JSONResponse({"page": "home", "datetime": str(datetime.utcnow())})
@app.get("/b/welcome")
@cache(key="b.home", end_of_week=True)
async def home(request: Request, response: Response):
return JSONResponse({"page": "welcome", "datetime": str(datetime.utcnow())})
While it's always possible to explicitly pass keys onto the key
attribute, there are scenarios where the keys need to be built based on the parameters received by the controller method. For instance, in an authenticated API where the user_id
is fetched as a controller Depends
argument.
@app.get("/b/logged-in")
@cache(key="b.logged_in.{}", obj="user", obj_attr="id")
async def logged_in(request: Request, response: Response, user=user):
return JSONResponse(
{"page": "home", "user": user.id, "datetime": str(datetime.utcnow())}
)
In the example above, the key allows room for a dynamic attribute fetched from the object user
. The key eventually becomes b.logged_in.112358
if the user.id
returns 112358
The cache invalidation can be managed using the @invalidate_cache
decorator.
@app.post("/b/logged-in")
@invalidate_cache(
key="b.logged_in.{}", obj="user", obj_attr="id", namespace="test_namespace"
)
async def post_logged_in(request: Request, response: Response, user=user):
return JSONResponse(
{"page": "home", "user": user.id, "datetime": str(datetime.utcnow())}
)
The cache invalidation decorator allows for multiple keys to be invalidated in the same call. However, the it assumes that the object attributes generated apply all keys.
@app.post("/b/logged-in")
@invalidate_cache(
keys=["b.logged_in.{}", "b.profile.{}"], obj="user", obj_attr="id", namespace="test_namespace"
)
async def post_logged_in(request: Request, response: Response, user=user):
return JSONResponse(
{"page": "home", "user": user.id, "datetime": str(datetime.utcnow())}
)
import os
import redis
from datetime import datetime
from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse
from c27cache import C27Cache, cache, invalidate_cache
REDIS_URL = os.environ.get("REDIS_URL", "redis://localhost:6379/3")
redis_client = redis.Redis.from_url(REDIS_URL)
class User:
id: str = "112358"
user = User()
app = FastAPI()
C27Cache.init(redis_url=REDIS_URL, namespace='test_namespace')
@app.get("/b/home")
@cache(key="b.home", ttl_in_seconds=180)
async def home(request: Request, response: Response):
return JSONResponse({"page": "home", "datetime": str(datetime.utcnow())})
@app.get("/b/logged-in")
@cache(key="b.logged_in.{}", obj="user", obj_attr="id")
async def logged_in(request: Request, response: Response, user=user):
return JSONResponse(
{"page": "home", "user": user.id, "datetime": str(datetime.utcnow())}
)
@app.post("/b/logged-in")
@invalidate_cache(
key="b.logged_in.{}", obj="user", obj_attr="id", namespace="test_namespace"
)
async def post_logged_in(request: Request, response: Response, user=user):
return JSONResponse(
{"page": "home", "user": user.id, "datetime": str(datetime.utcnow())}
)
ttl
dynamically for cache keys using a Callable
A callable can be passed as part of the decorator to dynamically compute what the ttl for a cache key should be. For example
async def my_ttl_callable() -> int:
return 3600
@app.get('/b/ttl_callable')
@cache(key='b.ttl_callable_expiry', ttl_func=my_ttl_callable)
async def path_with_ttl_callable(request: Request, response: Response):
return JSONResponse(
{"page": "path_with_ttl_callable", "datetime": str(datetime.utcnow())}
)
The ttl_func
is always assumed to be an async method
C27Cache works exactly the same way with regular methods. The example below explains usage of the cache in service objects and application services.
import os
import redis
from c27cache import C27Cache, cache, invalidate_cache
REDIS_URL = os.environ.get("REDIS_URL", "redis://localhost:6379/3")
redis_client = redis.Redis.from_url(REDIS_URL)
class User:
id: str = "112358"
user = User()
C27Cache.init(redis_url=REDIS_URL, namespace='test_namespace')
@cache(key='cache.me', ttl_in_seconds=360)
async def cache_me(x:int, invoke_count:int):
invoke_count = invoke_count + 1
result = x * 2
return [result, invoke_count]