codekingpro/portable-devtools

1"""Key/value cache for use inside LangGraph deployments.2 3Thin wrapper around ``langgraph_api.cache``.4Values must be JSON-serializable (dicts, lists, strings, numbers, booleans,5``None``).6"""7 8from __future__ import annotations9 10from collections.abc import Awaitable, Callable11from datetime import timedelta12from typing import Any, Generic, Literal, TypeVar13 14T = TypeVar("T")15 16CacheStatus = Literal["miss", "fresh", "stale", "expired"]17 18try:19    from langgraph_api.cache import (  # type: ignore[unresolved-import]20        cache_get as _cache_get,21    )22    from langgraph_api.cache import (  # type: ignore[unresolved-import]23        cache_set as _cache_set,24    )25except ImportError:26    _cache_get = None27    _cache_set = None28 29 30try:31    from langgraph_api.cache import SWRResult  # type: ignore[unresolved-import]32    from langgraph_api.cache import swr as _api_swr  # type: ignore[unresolved-import]33 34except ImportError:35    _api_swr = None36 37    class SWRResult(Generic[T]):38        """Result wrapper returned by :func:`swr`."""39 40        value: T41        status: CacheStatus42 43        async def mutate(self, value: T = ...) -> T:  # type: ignore[assignment]44            """Update or revalidate the cached value."""45            ...46 47 48__all__ = [49    "SWRResult",50    "cache_get",51    "cache_set",52    "swr",53]54 55 56async def cache_get(key: str) -> Any | None:57    """Get a value from the cache.58 59    Returns the deserialized value, or ``None`` if the key is missing or expired.60 61    Requires Agent Server runtime version 0.7.29 or later.62    """63    if _cache_get is None:64        raise RuntimeError(65            "Cache is only available server-side within the LangGraph Agent Server "66            "(https://docs.langchain.com/langsmith/deployments)."67        )68    return await _cache_get(key)69 70 71async def cache_set(key: str, value: Any, *, ttl: timedelta | None = None) -> None:72    """Set a value in the cache.73 74    Args:75        key: The cache key.76        value: The value to cache (must be JSON-serializable).77        ttl: Optional time-to-live. Capped at 1 day; ``None`` or zero78            defaults to 1 day.79 80    Requires Agent Server runtime version 0.7.29 or later.81    """82    if _cache_set is None:83        raise RuntimeError(84            "Cache is only available server-side within the LangGraph Agent Server "85            "(https://docs.langchain.com/langsmith/deployments)."86        )87    await _cache_set(key, value, ttl)88 89 90async def swr(91    key: str,92    loader: Callable[[], Awaitable[T]],93    *,94    fresh_for: timedelta | None = None,95    max_age: timedelta | None = None,96    model: type[T] | None = None,97) -> SWRResult[T]:98    """Load a cached value using stale-while-revalidate semantics.99 100    This helper is server-side only and is intended for caching internal async101    dependencies such as auth or metadata lookups.102 103    Args:104        key: Cache key.105        loader: Async callable that fetches the value on miss/revalidation.106        fresh_for: How long a cached value is considered fresh (no revalidation).107            Defaults to ``timedelta(0)`` so every access triggers a background108            revalidate while still returning the cached value instantly. Values109            above :data:`MAX_CACHE_TTL` are clamped to the backend maximum.110        max_age: Total lifetime of a cached entry. After this, the next access111            blocks on the loader. Defaults to :data:`MAX_CACHE_TTL` (24 h by112            default). Values above :data:`MAX_CACHE_TTL` are clamped to the113            backend maximum.114        model: Optional Pydantic model class. When provided, values are115            serialized via ``model_dump(mode="json")`` before storage and116            deserialized via ``model.model_validate()`` on read.117 118    Returns:119        An :class:`SWRResult` with ``.value``, ``.status``, and an async120        ``.mutate()`` method.121 122    Semantics:123    - cache miss: await ``loader()``, store the value, return it124    - fresh hit (age < fresh_for): return the cached value125    - stale hit (fresh_for <= age < max_age): return the cached value126      immediately and trigger a best-effort background refresh127    - expired (age >= max_age): await ``loader()``, store the value, return it128    """129    if _api_swr is None:130        raise RuntimeError(131            "Cache is only available server-side within the LangGraph Agent Server "132            "(https://docs.langchain.com/langsmith/deployments)."133        )134    if fresh_for is None:135        fresh_for = timedelta(0)136    if max_age is None:137        max_age = timedelta(days=1)138    return await _api_swr(139        key, loader, fresh_for=fresh_for, max_age=max_age, model=model140    )141