The @cached Decorator

The @cached decorator automatically caches function results. It works on async functions in either form, and on sync functions when you pass a TTLCache. A sync function runs off the event loop, from a worker thread such as asyncio.to_thread(...), and refuses to run on the loop that opened the cache. The ttl= form raises TypeError on a def, because the private cache it builds has no event loop for a sync wrapper.

For the plain "memoize this function for N seconds" case, pass ttl= and nothing else. The decorator builds a private process-local cache for this function alone:

from grelmicro.cache import cached

@cached(ttl=30)
async def get_rates() -> Rates:
    return await fetch_rates()

That private cache lives only in this process and is never shared across replicas. To share results across replicas, invalidate by tag, or reuse one store across functions, pass a TTLCache instead:

from grelmicro.cache import TTLCache, cached

cache = TTLCache[User](ttl=300)

@cached(cache)
async def get_user(user_id: int) -> User:
    return await db.fetch_user(user_id)

A TTLCache with no type parameter, such as TTLCache(ttl=300), caches a function that returns bytes or a JSON value like a float or a dict. See Serialization for the types each form stores.

Passing both cache and ttl, or neither, raises TypeError.

Custom Keys

By default @cached derives the key from the repr() of the arguments. Pass key_template= for a stable, readable key instead. The template fills in from the call's arguments, so key_template="user:{user_id}" keys the entry under user:42 for a call with user_id=42:

@cached(cache, key_template="user:{user_id}")
async def get_user(user_id: int) -> User:
    return await db.fetch_user(user_id)

Arguments not named in the template do not affect the key, so calls that differ only in those arguments share one entry. Defaults fill in when an argument is omitted. For a fully dynamic key, pass a key= function instead. It receives (func, args, kwargs) and returns the key. A string passed to key= raises TypeError, and so does passing both key and key_template. A custom key fully determines the lookup, so typed= has no effect when key= or key_template= is set.

fragment

@cached(cache, key=lambda func, args, kwargs: f"user:{args[0]}")
async def get_user(user_id: int) -> User:
    return await db.fetch_user(user_id)

key.py

import asyncio

from pydantic import BaseModel

from grelmicro import Grelmicro
from grelmicro.cache import cached
from grelmicro.providers.redis import RedisProvider

redis = RedisProvider("redis://localhost:6379/0")
micro = Grelmicro(uses=[redis])


class User(BaseModel):
    id: int
    name: str


ttl_cache = micro.cache.ttl(ttl=300, serializer=User)


@cached(ttl_cache, key_template="user:{user_id}")
async def get_user(user_id: int) -> User:
    return User(id=user_id, name="Alice")


async def main() -> None:
    async with micro:
        # Keyed under "user:1" instead of the default argument-repr key.
        print(await get_user(1))
        await ttl_cache.delete("user:1")


if __name__ == "__main__":
    asyncio.run(main())

Default keys are not stable across processes

The default key is the repr() of the arguments. Keys are stable within a single process but may vary across Python versions. An object using the default __repr__ carries a memory address, so its key changes on every restart and the entry is never found again. Pass key_template= or key= for such objects.

On a method

A method must name its key. Decorating one without key_template= or key= raises TypeError at decoration time:

class Repo:
    @cached(cache, key_template="repo:{user_id}")
    async def load(self, user_id: int) -> User:
        return await self.db.fetch_user(user_id)

The default key is the repr() of every argument, and on a method the first one is self. That reads two ways, both wrong. Two instances whose repr() matches share one entry, so a call on one returns the other's value. An instance using the default repr() carries a memory address, so its key changes on every restart and the entry is never found again.

Only you know what identifies the entry, so the decorator asks rather than guesses. Name the fields that matter and leave self out. When the result does depend on instance state, fold that state into the key:

class Repo:
    @cached(cache, key_template="repo:{self.region}:{user_id}")
    async def load(self, user_id: int) -> User: ...

A staticmethod and a classmethod are untouched. Neither receives an instance, and a class repr() is stable, so their default key is sound.

refresh() on a method is reached through the class, because attribute access on an instance returns the helper unbound:

fragment

await Repo.load.refresh(repo, 42)

On-demand refresh

refresh() recomputes for the given arguments, overwrites the stored entry, and returns the new value. It skips the read, so a live entry is replaced rather than served.

refresh.py

import asyncio

from pydantic import BaseModel

from grelmicro import Grelmicro
from grelmicro.cache import TTLCache, cached
from grelmicro.providers.redis import RedisProvider

redis = RedisProvider("redis://localhost:6379/0")
micro = Grelmicro(uses=[redis])


class Report(BaseModel):
    user_id: int


cache = TTLCache[Report](ttl=300)


@cached(cache, key_template="report:{user_id}")
async def get_report(user_id: int) -> Report:
    return Report(user_id=user_id)


async def main() -> None:
    async with micro:
        print(await get_report(42))  # computed, then stored
        print(await get_report(42))  # served from the cache
        # Skips the read, recomputes, and overwrites the entry.
        print(await get_report.refresh(42))


if __name__ == "__main__":
    asyncio.run(main())

Use it where a caller asks for fresh data, such as an endpoint honouring Cache-Control: no-cache:

@app.post("/reports/{user_id}")
async def report(user_id: int, cache_control: Annotated[str, Header()] = "") -> Report:
    if "no-cache" in cache_control:
        return await get_report.refresh(user_id)
    return await get_report(user_id)

Two things differ from a normal miss:

Refreshes do not fold. A miss folds concurrent callers into one execution, but every refresh runs the function itself. A refresh never returns a value computed before the call started, so writing to the database and then refreshing cannot hand you pre-write data. Under the default lock, refreshes for one key also serialize within the process, and with lock="cluster" they serialize across replicas. Under lock=None they run in parallel and the last write wins.
Errors propagate, even under stale_ttl. Serving the stale value would return the exact entry the caller asked to bypass. Compose it yourself when you want that:

fragment

try:
    return await get_report.refresh(user_id)
except UpstreamError:
    return await get_report(user_id)

A refresh honours skip, tags, and the stale_ttl reserve, so the stored entry stays consistent with what a normal miss would have written. A result the skip predicate rejects is not stored, so the previous entry stays and keeps being served. On a sync function refresh() returns the value directly, matching the call it mirrors.

Refreshing a key that was never cached simply computes and stores it. A refresh reads nothing, so it counts as neither a hit nor a miss in cache_info(). It works on the private @cached(ttl=...) form too.

Refreshing one key, not the whole cache

cache_clear() is bound to the whole TTLCache, so on a cache shared between functions it removes every function's entries. refresh() touches one key.

Streaming producers

Decorate an async generator and @cached caches what it yields. Iterating the result streams the items, and the assembled list is stored once the producer finishes.

stream.py

import asyncio
from collections.abc import AsyncIterator

from grelmicro import Grelmicro
from grelmicro.cache import JsonSerializer, TTLCache, cached
from grelmicro.providers.redis import RedisProvider

redis = RedisProvider("redis://localhost:6379/0")
micro = Grelmicro(uses=[redis])

cache = TTLCache(ttl=300, serializer=JsonSerializer())


@cached(cache, key_template="answer:{question_id}")
async def answer(question_id: int) -> AsyncIterator[str]:
    for token in ("The", " answer", " is", " 42."):
        yield token


async def main() -> None:
    async with micro:
        # Streams live from the producer, stores the tokens at the end.
        async for token in answer(1):
            print(token)
        # Replays the stored tokens, the producer does not run again.
        async for token in answer(1):
            print(token)
        # The same entry, read whole.
        tokens = await answer.collect(1)
        print("".join(tokens))


if __name__ == "__main__":
    asyncio.run(main())

This is the shape where one result is served two ways: a streaming endpoint that yields items as they are produced, and a buffered endpoint that returns the whole thing. Both read one entry, so whichever runs first pays for the work:

@app.get("/answer/{question_id}/stream")
async def stream(question_id: int) -> EventSourceResponse:
    return EventSourceResponse(answer(question_id))


@app.get("/answer/{question_id}")
async def whole(question_id: int) -> str:
    return "".join(await answer.collect(question_id))

Only a completed sequence is stored. A reader that stops early, and a producer that raises part way, both leave the key untouched. A truncated sequence is never published, so the next reader gets the whole thing rather than the part the first one happened to read.

A second reader waits, then replays. Under the default lock, a concurrent miss folds like any other: the second caller waits for the first to finish and then replays the stored entry, rather than running the producer again. It trades incremental output for not paying twice. Use lock=None to let both stream live at the cost of two executions.

The stored form is a plain list, so await cache.get(key) returns the items and anything else reading that key sees an ordinary cached value.

skip receives the assembled list, so skip=lambda items: not items declines to store an empty sequence. tags, early and refresh() work as they do anywhere else, and stale_ttl serves the reserve when the producer fails before its first item. Past that the caller already holds part of the live sequence, so the error propagates rather than replaying items it just read.

A sync generator is not supported and raises at decoration time. It yields its items once, so a cached one would replay as empty.

Stampede Protection

A cache stampede (or "dog-pile") happens when many callers miss the same key at once and all recompute it together. @cached folds those misses: they share a single call of the function. lock= says how far the fold reaches, and early= refreshes hot keys before they expire:

Setting	What it does	Cost	Use when
`lock="process"` (default)	fold misses in the process, never touches a backend	free, no I/O	the common case
`lock="host"`	fold misses in the process, then across the processes on one host	one lock backend call per cold miss	several workers share one host
`lock="cluster"`	fold misses in the process, then across every replica	one lock backend call per cold miss	each call is expensive enough to run once for the fleet
`lock=None`	no folding, every concurrent miss calls the function	none	misses are cheap or rare
`early=0.1`	probabilistic early refresh (XFetch) in the last 10% of the TTL	one background recompute per refresh	the hottest keys, where no caller should ever block

"process" folds inside one process only, so each replica still calls the function once. "host" and "cluster" always fold in the process first, so the lock backend is called at most once per cold miss. early= works with any lock. An early refresh is decided by each reader, so each replica refreshes a hot key on its own.

@cached(cache)                     # default: fold misses in the process
async def get_user(user_id: int) -> User:
    return await db.fetch_user(user_id)


@cached(cache, lock="cluster")     # fold misses across replicas
async def get_billing(user_id: int) -> Billing:
    return await billing.fetch(user_id)


@cached(cache, early=0.1)          # refresh hot keys before they expire
async def get_homepage_feed() -> Feed:
    return await build_feed()

lock is per-key: concurrent misses on different keys run in parallel. Only callers that request the same key wait in turn, so one slow computation does not block unrelated keys.

"host" and "cluster" fold through the lock backend of the app's Coordination:

The backend must reach that far. It goes through the same backend check as requires=. In staging and production, a miss on a lock backend whose scope falls short raises BackendScopeError. With no environment declared it logs a warning, and in development and test it stays quiet.
No lock backend raises OutOfContextError. Register a Coordination on the app, or keep lock="process".
The check runs when the app opens. A fold with no Coordination, or one whose lock backend falls short, is reported by micro.describe() and grelmicro check as a backend-scope warning before any call. The app warns with BackendScopeWarning on startup and still opens. Any app opened in the process reports a decorated fold, even one it never calls. Pass lock="process" or register a Coordination that reaches far enough to silence it.
A failing lock backend does not fail the call. The error is logged and the misses fold in the process only.
A lock lost before release is logged, for example when the function runs past the lease. The value is still returned.

@cached(ttl=...) keeps its entries in one process, so it accepts only "process" or None. lock="host" or lock="cluster" there raises SettingsValidationError. Pass a shared TTLCache to fold across processes.

early= returns the cached value immediately and recomputes in the background, so a hot key refreshes before it expires and no caller ever waits on a cold miss. It costs one extra recompute per refresh and stores a small sidecar entry next to the value so replicas coordinate the refresh window.

When to use: your cached function is expensive (database query, API call, heavy computation) and may be called concurrently with the same arguments.

Serve Stale on Error

Set stale_ttl to keep serving the last good value when a recompute fails. Each result is also kept as a fallback copy for ttl + stale_ttl. After the TTL, the next miss recomputes as usual, but if that recompute raises, the most recent value is served instead of propagating the error, for up to stale_ttl past the TTL.

cache = TTLCache[Rates](ttl=60)

@cached(cache, stale_ttl=600)
async def get_exchange_rates() -> Rates:
    return await rates_api.fetch()   # a flaky external call

A flaky upstream then degrades to slightly stale data instead of an error storm. Once the recompute succeeds again, the fresh value takes over. If the upstream stays down longer than stale_ttl, the error propagates.

stale_ttl composes with lock and early. An explicit cache.delete(...) or cache.delete_tags(...) drops the fallback too, so invalidation is never undone by a later stale serve. Each stale serve records the grelmicro.cache.stale_serves metric, so a rising count signals an unhealthy upstream.

Decorator Parameters

cache and ttl are mutually exclusive. Pass one or the other, not both.

Parameter	Type	Default	Description
`cache`	`TTLCache`	`None`	The cache instance to store results in. Mutually exclusive with `ttl`.
`ttl`	`int \\| timedelta`	`None`	TTL for a private per-function cache, in whole seconds or as a `timedelta`. Mutually exclusive with `cache`.
`maxsize`	`int`	`0`	Max entries in the private per-function cache, `0` means unlimited (used only when `ttl` is set).
`key`	`Callable`	`None`	Function deriving the key. Receives `(func, args, kwargs)`. Mutually exclusive with `key_template`.
`key_template`	`str`	`None`	Key template rendered from the arguments, like `"user:{user_id}"`. Mutually exclusive with `key`.
`skip`	`Callable`	`None`	Predicate receiving the result. Returns `True` to skip caching.
`typed`	`bool`	`False`	Cache arguments of different types separately.
`lock`	`"process"`, `"host"`, `"cluster"`, or `None`	`"process"`	How far concurrent misses on one key fold into one call.
`early`	`float` in `[0, 1)`	`None`	Probabilistic early refresh in the late TTL window.
`stale_ttl`	`int \\| timedelta`	`None`	Serve-stale-on-error budget, in whole seconds or as a `timedelta`. Serve the last good value for this long past the TTL when a recompute fails.
`tags`	`Sequence[str]`	`()`	Tags to attach to each result. Templates like `"user:{user_id}"` fill in from the arguments. Invalidate with `cache.delete_tags(...)`.

Decorated Function Helpers

Helper	Returns	Description
`refresh(args, *kwargs)`	the new value	Recompute for these arguments and overwrite the stored entry. On a method, call it as `Class.method.refresh(obj, ...)`.
`collect(args, *kwargs)`	awaitable list	Read the whole sequence of a streaming producer. Only on an async generator.
`cache_info()`	`CacheInfo`	Statistics for the whole cache backing this function.
`cache_clear()`	awaitable	Remove every entry from that cache. Always a coroutine, even for a sync function.

The @cached Decorator

Custom Keys

Tags

On a method

On-demand refresh

Streaming producers

Stampede Protection

Serve Stale on Error

Decorator Parameters

Decorated Function Helpers