codekingpro/portable-devtools

1from __future__ import annotations2 3from abc import ABC, abstractmethod4from collections.abc import Mapping, Sequence5from typing import Generic, TypeVar6 7from langgraph.checkpoint.serde.base import SerializerProtocol8from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer9 10ValueT = TypeVar("ValueT")11Namespace = tuple[str, ...]12FullKey = tuple[Namespace, str]13 14 15class BaseCache(ABC, Generic[ValueT]):16    """Base class for a cache."""17 18    serde: SerializerProtocol = JsonPlusSerializer(pickle_fallback=False)19 20    def __init__(self, *, serde: SerializerProtocol | None = None) -> None:21        """Initialize the cache with a serializer."""22        self.serde = serde or self.serde23 24    @abstractmethod25    def get(self, keys: Sequence[FullKey]) -> dict[FullKey, ValueT]:26        """Get the cached values for the given keys."""27 28    @abstractmethod29    async def aget(self, keys: Sequence[FullKey]) -> dict[FullKey, ValueT]:30        """Asynchronously get the cached values for the given keys."""31 32    @abstractmethod33    def set(self, pairs: Mapping[FullKey, tuple[ValueT, int | None]]) -> None:34        """Set the cached values for the given keys and TTLs."""35 36    @abstractmethod37    async def aset(self, pairs: Mapping[FullKey, tuple[ValueT, int | None]]) -> None:38        """Asynchronously set the cached values for the given keys and TTLs."""39 40    @abstractmethod41    def clear(self, namespaces: Sequence[Namespace] | None = None) -> None:42        """Delete the cached values for the given namespaces.43        If no namespaces are provided, clear all cached values."""44 45    @abstractmethod46    async def aclear(self, namespaces: Sequence[Namespace] | None = None) -> None:47        """Asynchronously delete the cached values for the given namespaces.48        If no namespaces are provided, clear all cached values."""49