mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-08-28 10:49:56 +02:00
Sydney RunkleandGitHub
228a08b966
* Migrate to `uv` * Format `pyproject.toml` files properly * Remove upper bounds on dependencies, and bounds on dev dependencies (we should be using latest) * Move to hatch for packaing In the future we should: * Set up dependabot / automate lockfile updates and tests * Add tests for min compatible versions (I'll do this right after merge) * Use dynamic versioning * Bump `pydantic` to v2.11.4 in the lockfile, we have some tests failing
49 lines
1.8 KiB
Python
49 lines
1.8 KiB
Python
from __future__ import annotations
|
|
|
|
from abc import ABC, abstractmethod
|
|
from collections.abc import Mapping, Sequence
|
|
from typing import Generic, TypeVar
|
|
|
|
from langgraph.checkpoint.serde.base import SerializerProtocol
|
|
from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
|
|
|
|
ValueT = TypeVar("ValueT")
|
|
Namespace = tuple[str, ...]
|
|
FullKey = tuple[Namespace, str]
|
|
|
|
|
|
class BaseCache(ABC, Generic[ValueT]):
|
|
"""Base class for a cache."""
|
|
|
|
serde: SerializerProtocol = JsonPlusSerializer(pickle_fallback=True)
|
|
|
|
def __init__(self, *, serde: SerializerProtocol | None = None) -> None:
|
|
"""Initialize the cache with a serializer."""
|
|
self.serde = serde or self.serde
|
|
|
|
@abstractmethod
|
|
def get(self, keys: Sequence[FullKey]) -> dict[FullKey, ValueT]:
|
|
"""Get the cached values for the given keys."""
|
|
|
|
@abstractmethod
|
|
async def aget(self, keys: Sequence[FullKey]) -> dict[FullKey, ValueT]:
|
|
"""Asynchronously get the cached values for the given keys."""
|
|
|
|
@abstractmethod
|
|
def set(self, pairs: Mapping[FullKey, tuple[ValueT, int | None]]) -> None:
|
|
"""Set the cached values for the given keys and TTLs."""
|
|
|
|
@abstractmethod
|
|
async def aset(self, pairs: Mapping[FullKey, tuple[ValueT, int | None]]) -> None:
|
|
"""Asynchronously set the cached values for the given keys and TTLs."""
|
|
|
|
@abstractmethod
|
|
def clear(self, namespaces: Sequence[Namespace] | None = None) -> None:
|
|
"""Delete the cached values for the given namespaces.
|
|
If no namespaces are provided, clear all cached values."""
|
|
|
|
@abstractmethod
|
|
async def aclear(self, namespaces: Sequence[Namespace] | None = None) -> None:
|
|
"""Asynchronously delete the cached values for the given namespaces.
|
|
If no namespaces are provided, clear all cached values."""
|