Real-World Examples
===================

FastAPI
-------

.. code-block:: python

    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
-----------------------

.. code-block:: python

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

.. code-block:: python

    # 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
-------

.. code-block:: python

    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:

.. code-block:: python

    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
-----------------

.. code-block:: python

    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
----------------

.. code-block:: python

    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)