Skip to content

API

pico_caching.decorators

@cacheable marker: stores the policy and attaches the interceptor.

cacheable(_func=None, *, ttl_seconds=None, key=None)

Cache the method's return value. Sync or async; async methods cache the awaited result, never the coroutine. key(*args, **kwargs) overrides the default repr-based key; ttl_seconds overrides cache.default_ttl_seconds.

Source code in src/pico_caching/decorators.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def cacheable(
    _func: Callable | None = None,
    *,
    ttl_seconds: float | None = None,
    key: Callable[..., str] | None = None,
):
    """Cache the method's return value. Sync or async; async methods cache
    the awaited result, never the coroutine. ``key(*args, **kwargs)``
    overrides the default repr-based key; ``ttl_seconds`` overrides
    ``cache.default_ttl_seconds``."""

    def dec(fn):
        setattr(fn, CACHE_META, {"ttl_seconds": ttl_seconds, "key": key})
        from pico_ioc import intercepted_by

        from .interceptor import CacheInterceptor

        return intercepted_by(CacheInterceptor)(fn)

    return dec(_func) if callable(_func) else dec

pico_caching.interceptor

Cache interceptor: get-or-compute through the selected backend.

CacheInterceptor

Uses the first user-provided CacheBackend; falls back to the built-in in-memory LRU.

Source code in src/pico_caching/interceptor.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
@component(scope="singleton")
class CacheInterceptor:
    """Uses the first user-provided ``CacheBackend``; falls back to the
    built-in in-memory LRU."""

    def __init__(self, settings: CacheSettings, backends: List[CacheBackend]):
        self.settings = settings
        custom = [b for b in backends if not isinstance(b, InMemoryCacheBackend)]
        fallback = [b for b in backends if isinstance(b, InMemoryCacheBackend)]
        self.backend = (custom or fallback)[0]

    def invoke(self, ctx: MethodCtx, call_next: Callable[[MethodCtx], Any]) -> Any:
        meta = _meta(ctx)
        if not self.settings.enabled or not meta:
            return call_next(ctx)
        key_fn = meta["key"]
        key = key_fn(*ctx.args, **ctx.kwargs) if key_fn else _default_key(ctx)
        ttl = meta["ttl_seconds"] if meta["ttl_seconds"] is not None else self.settings.default_ttl_seconds
        if inspect.iscoroutinefunction(ctx.method):
            return self._get_or_compute_async(ctx, call_next, key, ttl)
        hit, value = self.backend.get(key)
        if hit:
            return value
        value = call_next(ctx)
        self.backend.set(key, value, ttl)
        return value

    async def _get_or_compute_async(self, ctx, call_next, key: str, ttl: float):
        hit, value = self.backend.get(key)
        if hit:
            return value
        value = await call_next(ctx)
        self.backend.set(key, value, ttl)
        return value

pico_caching.backend

Cache backend protocol and the built-in in-memory LRU backend.

CacheBackend

Bases: Protocol

Implement as a @component to replace the in-memory backend.

Source code in src/pico_caching/backend.py 
15
16
17
18
19
20
21
22
@runtime_checkable
class CacheBackend(Protocol):
    """Implement as a ``@component`` to replace the in-memory backend."""

    def get(self, key: str) -> Tuple[bool, Any]: ...
    def set(self, key: str, value: Any, ttl_seconds: float) -> None: ...
    def delete(self, key: str) -> None: ...
    def clear(self) -> None: ...

InMemoryCacheBackend

Thread-safe LRU with per-entry TTL (time.monotonic based).

Source code in src/pico_caching/backend.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
@component
class InMemoryCacheBackend:
    """Thread-safe LRU with per-entry TTL (``time.monotonic`` based)."""

    def __init__(self, settings: CacheSettings):
        self._max = settings.max_entries
        self._data: OrderedDict[str, Tuple[float, Any]] = OrderedDict()
        self._lock = threading.Lock()

    def get(self, key: str) -> Tuple[bool, Any]:
        with self._lock:
            item = self._data.get(key, _MISS)
            if item is _MISS:
                return False, None
            expires_at, value = item
            if time.monotonic() >= expires_at:
                del self._data[key]
                return False, None
            self._data.move_to_end(key)
            return True, value

    def set(self, key: str, value: Any, ttl_seconds: float) -> None:
        with self._lock:
            self._data[key] = (time.monotonic() + ttl_seconds, value)
            self._data.move_to_end(key)
            while len(self._data) > self._max:
                self._data.popitem(last=False)

    def delete(self, key: str) -> None:
        with self._lock:
            self._data.pop(key, None)

    def clear(self) -> None:
        with self._lock:
            self._data.clear()

pico_caching.config

Settings for pico-caching (prefix caching, zero-config).

CacheSettings dataclass

enabled: false makes @cacheable a pass-through.

Source code in src/pico_caching/config.py 
 8
 9
10
11
12
13
14
15
@configured(target="self", prefix="caching", mapping="tree")
@dataclass
class CacheSettings:
    """``enabled: false`` makes ``@cacheable`` a pass-through."""

    enabled: bool = True
    default_ttl_seconds: float = 300.0
    max_entries: int = 1024