From c937d5f04863fb0f7f8c417d8a57ccb3a0738bb0 Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Thu, 8 May 2025 16:54:06 -0700 Subject: [PATCH] Lint --- .../langgraph/cache/file/__init__.py | 117 ------------------ libs/checkpoint/langgraph/cache/file/py.typed | 0 .../langgraph/cache/memory/__init__.py | 10 +- 3 files changed, 5 insertions(+), 122 deletions(-) delete mode 100644 libs/checkpoint/langgraph/cache/file/__init__.py delete mode 100644 libs/checkpoint/langgraph/cache/file/py.typed diff --git a/libs/checkpoint/langgraph/cache/file/__init__.py b/libs/checkpoint/langgraph/cache/file/__init__.py deleted file mode 100644 index 29231b206..000000000 --- a/libs/checkpoint/langgraph/cache/file/__init__.py +++ /dev/null @@ -1,117 +0,0 @@ -from __future__ import annotations - -import asyncio -import datetime -import sqlite3 -import threading -from collections.abc import Mapping, Sequence - -from langgraph.cache.base import BaseCache, FullKey, Namespace -from langgraph.checkpoint.serde.base import SerializerProtocol - - -class FileCache(BaseCache): - """File-based cache using SQLite.""" - - def __init__( - self, - *, - path: str, - serde: SerializerProtocol | None = None, - ) -> None: - """Initialize the cache with a file path.""" - super().__init__(serde=serde) - # SQLite backing store - self._conn = sqlite3.connect( - path, - check_same_thread=False, - ) - # Serialize access to the shared connection across threads - self._lock = threading.RLock() - # Better concurrency & atomicity - self._conn.execute("PRAGMA journal_mode=WAL;") - # Schema: key -> (expiry, encoding, value) - self._conn.execute( - """CREATE TABLE IF NOT EXISTS cache ( - ns TEXT, - key TEXT, - expiry REAL, - encoding TEXT NOT NULL, - val BLOB NOT NULL, - PRIMARY KEY (ns, key) - )""" - ) - self._conn.commit() - - def get(self, keys: Sequence[FullKey]) -> dict[FullKey, bytes]: - """Get the cached values for the given keys.""" - with self._lock, self._conn: - now = datetime.datetime.now(datetime.timezone.utc).timestamp() - if not keys: - return {} - placeholders = ",".join("(?, ?)" for _ in keys) - params: list[str] = [] - for ns_tuple, key in keys: - params.extend((",".join(ns_tuple), key)) - cursor = self._conn.execute( - f"SELECT ns, key, expiry, encoding, val FROM cache WHERE (ns, key) IN ({placeholders})", - tuple(params), - ) - values: dict[FullKey, bytes] = {} - rows = cursor.fetchall() - for ns, key, expiry, encoding, raw in rows: - if expiry is not None and now > expiry: - # purge expired entry - self._conn.execute( - "DELETE FROM cache WHERE (ns, key) = (?, ?)", (ns, key) - ) - continue - values[(tuple(ns.split(",")), key)] = self.serde.loads_typed( - (encoding, raw) - ) - return values - - async def aget(self, keys: Sequence[FullKey]) -> dict[FullKey, bytes]: - """Asynchronously get the cached values for the given keys.""" - return await asyncio.to_thread(self.get, keys) - - def set(self, mapping: Mapping[FullKey, tuple[bytes, int | None]]) -> None: - """Set the cached values for the given keys and TTLs.""" - with self._lock, self._conn: - now = datetime.datetime.now(datetime.timezone.utc) - for key, (value, ttl) in mapping.items(): - if ttl is not None: - delta = datetime.timedelta(seconds=ttl) - expiry: float | None = (now + delta).timestamp() - else: - expiry = None - encoding, raw = self.serde.dumps_typed(value) - self._conn.execute( - "INSERT OR REPLACE INTO cache (ns, key, expiry, encoding, val) VALUES (?, ?, ?, ?, ?)", - (",".join(key[0]), key[1], expiry, encoding, raw), - ) - - async def aset(self, mapping: Mapping[FullKey, tuple[bytes, int | None]]) -> None: - """Asynchronously set the cached values for the given keys and TTLs.""" - await asyncio.to_thread(self.set, mapping) - - def delete(self, keys: Sequence[Namespace]) -> None: - """Delete the cached values for the given namespaces.""" - if not keys: - return - with self._lock, self._conn: - placeholders = ",".join("?" for _ in keys) - self._conn.execute( - f"DELETE FROM cache WHERE (ns) IN ({placeholders})", - tuple(",".join(key) for key in keys), - ) - - async def adelete(self, keys: Sequence[Namespace]) -> None: - """Asynchronously delete the cached values for the given namespaces.""" - await asyncio.to_thread(self.delete, keys) - - def __del__(self) -> None: - try: - self._conn.close() - except Exception: - pass diff --git a/libs/checkpoint/langgraph/cache/file/py.typed b/libs/checkpoint/langgraph/cache/file/py.typed deleted file mode 100644 index e69de29bb..000000000 diff --git a/libs/checkpoint/langgraph/cache/memory/__init__.py b/libs/checkpoint/langgraph/cache/memory/__init__.py index d93744e80..c39d4b40f 100644 --- a/libs/checkpoint/langgraph/cache/memory/__init__.py +++ b/libs/checkpoint/langgraph/cache/memory/__init__.py @@ -2,7 +2,7 @@ from __future__ import annotations import datetime import threading -from collections.abc import Sequence +from collections.abc import Mapping, Sequence from typing import Generic from langgraph.cache.base import BaseCache, FullKey, Namespace, ValueT @@ -12,7 +12,7 @@ from langgraph.checkpoint.serde.base import SerializerProtocol class InMemoryCache(BaseCache[ValueT], Generic[ValueT]): def __init__(self, *, serde: SerializerProtocol | None = None): super().__init__(serde=serde) - self._cache: dict[Namespace, dict[str, tuple[str, bytes, int | None]]] = {} + self._cache: dict[Namespace, dict[str, tuple[str, bytes, float | None]]] = {} self._lock = threading.RLock() def get(self, keys: Sequence[FullKey]) -> dict[FullKey, ValueT]: @@ -36,10 +36,10 @@ class InMemoryCache(BaseCache[ValueT], Generic[ValueT]): """Asynchronously get the cached values for the given keys.""" return self.get(keys) - def set(self, keys: dict[FullKey, tuple[ValueT, int | None]]) -> None: + def set(self, keys: Mapping[FullKey, tuple[ValueT, int | None]]) -> None: """Set the cached values for the given keys.""" with self._lock: - now = datetime.datetime.now(datetime.timezone.utc).timestamp() + now = datetime.datetime.now(datetime.timezone.utc) for (ns, key), (value, ttl) in keys.items(): if ttl is not None: delta = datetime.timedelta(seconds=ttl) @@ -53,7 +53,7 @@ class InMemoryCache(BaseCache[ValueT], Generic[ValueT]): expiry, ) - async def aset(self, keys: dict[FullKey, tuple[ValueT, int | None]]) -> None: + async def aset(self, keys: Mapping[FullKey, tuple[ValueT, int | None]]) -> None: """Asynchronously set the cached values for the given keys.""" self.set(keys)