Real-World Examples

FastAPI

from contextlib import asynccontextmanager
from fastapi import FastAPI
from cache import AsyncCache, DiskBackend

backend = DiskBackend("/tmp/fastapi_cache.pkl")
cache = AsyncCache(maxsize=5000, default_ttl=300, backend=backend)

@asynccontextmanager
async def lifespan(app):
    # Startup: cache loads from disk automatically via backend
    await cache.warmup({
        "config:app": load_config,
    })
    yield
    # Shutdown: persist to disk
    cache.save_to_backend()

app = FastAPI(lifespan=lifespan)

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    return await cache.get(
        f"user:{user_id}",
        loader=lambda: db.get_user(user_id),
        ttl=60,
    )

@app.put("/users/{user_id}")
async def update_user(user_id: int, data: dict):
    await db.update_user(user_id, data)
    cache.delete(f"user:{user_id}")
    return {"status": "updated"}

@app.get("/metrics")
async def metrics():
    return cache.get_metrics()

FastAPI with Decorators

from cache import AsyncTTL, AsyncTTLInvalidator

@AsyncTTL(time_to_live=120)
async def get_product(product_id: int):
    return await db.get_product(product_id)

@AsyncTTLInvalidator(get_product, key_fn=lambda a, kw: (a[:1], {}))
async def update_product(product_id: int, data: dict):
    await db.update_product(product_id, data)

@app.get("/products/{pid}")
async def product_endpoint(pid: int):
    return await get_product(pid)

@app.put("/products/{pid}")
async def update_product_endpoint(pid: int, data: dict):
    await update_product(pid, data)
    return {"status": "updated"}

Django (async views)

# views.py
from cache import AsyncCache, DiskBackend
from django.http import JsonResponse

backend = DiskBackend("/var/cache/django_app/cache.pkl")
cache = AsyncCache(maxsize=10000, default_ttl=600, backend=backend)

async def user_view(request, user_id):
    user = await cache.get(
        f"user:{user_id}",
        loader=lambda: User.objects.aget(pk=user_id),
    )
    return JsonResponse({"user": user})

# In AppConfig.ready() or management command:
# cache.save_to_backend()  # on shutdown

aiohttp

from aiohttp import web
from cache import AsyncCache, DiskBackend

backend = DiskBackend("/tmp/aiohttp_cache.pkl")
cache = AsyncCache(maxsize=5000, default_ttl=300, backend=backend)

async def get_user(request):
    user_id = request.match_info["user_id"]
    user = await cache.get(
        f"user:{user_id}",
        loader=lambda: db.get_user(user_id),
    )
    return web.json_response(user)

async def on_startup(app):
    await cache.warmup({"config": load_config})

async def on_shutdown(app):
    cache.save_to_backend()

app = web.Application()
app.router.add_get("/users/{user_id}", get_user)
app.on_startup.append(on_startup)
app.on_shutdown.append(on_shutdown)

ML Inference Cache

Expensive model inference results persisted across restarts:

from cache import AsyncLRU, DiskBackend

model_backend = DiskBackend("/var/cache/ml/embeddings.pkl")

@AsyncLRU(maxsize=50000, backend=model_backend)
async def get_embedding(text: str):
    return await model.encode(text)  # expensive GPU call

# On shutdown
import atexit
atexit.register(get_embedding.save_to_backend)

# On restart, cache is warm from disk — no cold-start spike

Web Scraper Cache

from cache import AsyncTTL, DiskBackend

scraper_backend = DiskBackend("/tmp/scraper_cache.pkl")

@AsyncTTL(time_to_live=3600, maxsize=100000, backend=scraper_backend)
async def fetch_page(url: str):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            return await resp.text()

# Pages cached for 1 hour, survive restarts
html = await fetch_page("https://example.com/data")

E-Commerce Agent

from agent_cache import AgentCache, AgentCacheInvalidator, AgentCacheSession

@AgentCache(resource="cart", scope="global", ttl=120)
async def get_cart(user_id):
    return await shop_api.get_cart(user_id)

@AgentCacheInvalidator(resource="cart", scope="global")
async def add_to_cart(user_id, item):
    return await shop_api.add_to_cart(user_id, item)

async with AgentCacheSession(loop_detection=False) as session:
    cart = await get_cart("user_1")         # DB fetch
    await add_to_cart("user_1", "laptop")   # invalidates cart
    cart = await get_cart("user_1")         # re-fetched (correct)