tagged-cache 0.1.0

A flexible async/sync Redis cache with tag-based invalidation support for FastAPI, Pydantic

Tagged Cache

A flexible async/sync Redis cache with tag-based invalidation support for FastAPI, Pydantic

Installation

pip install tagged-cache

Example

import asyncio
from fastapi import FastAPI
from redis.asyncio import Redis
from pydantic import BaseModel
from tagged_cache import TaggedCache
from contextlib import asynccontextmanager

redis_client = Redis(host="localhost", port=6379, decode_responses=True)
cache = TaggedCache(redis_client=redis_client)


@asynccontextmanager
async def lifespan(_: FastAPI):
    yield
    redis_client.close()

app = FastAPI(title="TaggedCache Base Test", lifespan=lifespan)


class TestModel(BaseModel):
    message: str
    counter: int

request_counter = 0

@app.get("/test-cache/{item_id}", response_model=TestModel)
@cache.cache(ttl=60, tags=["item:{item_id}"])
async def get_cached_item(item_id: int):
    global request_counter
    request_counter += 1

    await asyncio.sleep(1.0)
    
    return TestModel(
        message=f"Item {item_id} data fetched from heavy database", 
        counter=request_counter
    )

@app.post("/test-invalidate/{item_id}")
async def invalidate_item_cache(item_id: int):
    await cache.invalidate_tag(f"item:{item_id}")
    return {"status": f"Cache for item:{item_id} successfully invalidated"}