Python cache library for use with asyncio. This including retries (tenacity), exponential backoff and serialization. It is possible add more backends and serializers easily. Require Python 3.10+
- Memory
- Redis (https://github.com/redis/redis-py)
- pickle (default)
- msgpack
- orjson
Null Cache is the default backend. Your app will run without cache.
from aio_cache.cache_service import CacheService
cache = CacheService()
await cache.set("hello", "world")
from aio_cache.cache_service import CacheService
cache = CacheService()
cache.initialize(cache_uri="redis://localhost/0?ttl=60", prefix="my_prefix")
await cache.set("hello", [1, 2, 3])
result = await cache.get("hello")
await cache.mset([("key1", 1), ("key2": "bye")]) # multi set using a pipeline
results = await cache.mget(["key1", "key2"]) # multi get using a pipeline
await cache.delete_by_pattern("key") # delete keys thats match with "key*"
from aio_cache.cache_service import CacheService
cache = CacheService()
cache.initialize(cache_uri="memory://")
await cache.set("hello", "world", ttl=60)
from aio_cache.cache_service import CacheService
cache = CacheService()
cache.initialize(cache_uri="memory://", serializer="json")
await cache.set("hello", "world")
You can define a custom exception for handle when an error appears
class CustomCacheException(Exception):
pass
cache = CacheService(cache_error_exception=CustomCacheException)
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from aio_cache.cache_service import CacheService
class RedisException(Exception):
def __init__(self, name: str):
self.name = name
app = FastAPI()
cache = CacheService(cache_error_exception=RedisException)
cache.initialize(cache_uri="redis://localhost/0", prefix="fast_")
@app.exception_handler(RedisException)
async def redis_exception_handler(request: Request, exc: RedisException):
return JSONResponse(
status_code=418,
content={"message": f"Oops! {exc.name}. Maybe you must to check your redis..."},
)
@app.get("/cache/{name}/")
async def read_name(name: str):
result = await cache.get(name)
return {"result": result}