GitHub - grandimam/zinda: Freshness-first caching library for FastAPI

Experimental — This library is being actively designed. Core ideas are still being validated. Do not use in production. Feedback, experiments, and critical review are welcome.

Freshness-first caching for FastAPI. The goal is automatic caching that adapts to your hot paths instead of forcing you to tune every TTL and invalidation rule by hand.

Quickstart (FastAPI)

pip install -e ".[fastapi]"

from fastapi import FastAPI
from zinda import Cache
from zinda.ext.fastapi import install

app = FastAPI()
cache = install(app, Cache(default_ttl=60))

@cache.cached(ttl=120, refresh_after=30)
async def fetch_products(category: str, page: int = 1):
    return await db.fetch_products(category, page)

@app.get("/products/{category}")
async def products(category: str, page: int = 1):
    return await fetch_products(category, page)

That's it:

install(app, cache) starts the auto-refresh sweeper with your app's lifespan and exposes GET /zinda/stats so you can see every caching decision.
@cache.cached(...) caches the function's return value, keyed automatically on its arguments. Concurrent calls for the same arguments coalesce into one recompute (no stampedes).
ttl=120 is the hard expiry. refresh_after=30 is the soft expiry: after 30s, callers still get an instant (stale) response while zinda refreshes the value in the background — and the sweeper refreshes hot entries before anyone even asks.

Using it in a FastAPI app

1. Cache your data-access functions (recommended)

Works on any async function — service layer, repository, or a FastAPI dependency:

@cache.cached(ttl=300)
async def dashboard_stats():
    return await db.heavy_aggregation()

@app.get("/dashboard")
async def dashboard(stats=Depends(dashboard_stats)):
    return stats

Sync functions work too, they run in a thread:

@cache.cached(ttl=60)
def expensive_calculation(x: int) -> int:
    return x ** x

2. Cache whole responses with middleware (zero-touch)

Coarse but requires no changes to your endpoints: caches raw HTTP 200 responses for GETs, keyed by method + path + query string.

from zinda.ext.fastapi import CacheMiddleware

app.add_middleware(CacheMiddleware, cache=cache, ttl=60, refresh_after=30)

Non-GET methods, non-200 responses, and /zinda/* paths are never cached.

3. See what it's doing

curl localhost:8000/zinda/stats

{
  "fetch_products": {
    "calls": 142,
    "hits": 120,
    "misses": 22,
    "coalesced": 15,
    "stale_serves": 8,
    "refreshes": 6,
    "refresh_errors": 0,
    "hit_rate": 0.845,
    "avg_miss_ms": 48.3,
    "calls_per_second": 2.4
  }
}

Same data programmatically: cache.report().

Beyond FastAPI

zinda works anywhere async Python runs:

from zinda import Cache

cache = Cache(default_ttl=60)

# direct access
await cache.set("config:flags", {"beta": True}, ttl=30)
flags = await cache.get("config:flags")

# read-through
profile = await cache.get_or_set("user:42", lambda: db.fetch_user(42), ttl=120)

# run the sweeper yourself (install() does this for you in FastAPI)
await cache.start_autorefresh()
...
await cache.stop_autorefresh()

Escape hatches

Arguments that can't be stably represented (plain objects, sets) raise KeyDerivationError instead of silently caching wrong. Provide your own key:

@cache.cached(key_builder=lambda req: f"feed:{req.user_id}:{req.cursor}")
async def get_feed(req: FeedRequest):
    return await build_feed(req)

Status

Phase	Feature	Status
P0	Decorator API, in-memory backend, TTL, single-flight coalescing, stats	Implemented and tested
P1	Hotpath analyzer: auto-detect what to cache	In design
P2	Adaptive TTL: learn freshness per keyspace	In design
P3	Tag-based invalidation	Not started
P4	FastAPI middleware + lifecycle integration	Implemented, not hardened
P5	Redis backend + tiered L1/L2	Not started

Development

pip install -e ".[dev]"
python3 -m pytest