c27cache 0.1.6

A simple caching library.

C27Cache

C27Cache is a simple HTTP caching library designed to work with FastAPI

Installation

While C27Cache is still early in it's development, it's been used in production on a couple of services.

With pip

pip install c27cache

With Poetry

poetry add c27cache

Usage and Examples

Basic Usage

1. Initialize 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.

2. Define your controllers

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())})    

Building keys from parameter objects.

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

Explicitly invalidating the cache

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())}
    )

Invalidating more than one key at a time

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())}
    )

Full Example

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())}
    )

Computing 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

Caching methods that aren't controllers

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]