Overview
A file-based cache implementation using SQLite that provides key-value storage with optional TTL (time-to-live) expiration for LangGraph operations.
Description
The SqliteCache class extends `BaseCache` to provide a persistent, file-based caching layer backed by SQLite. It stores cached values in a single `cache` table with columns for namespace, key, expiry timestamp, encoding type, and serialized value data. The cache uses WAL (Write-Ahead Logging) journal mode for improved concurrency and atomicity, and employs a `threading.RLock` to serialize access across threads.
TTL support is implemented at the application level: when setting a value, the caller may optionally provide a TTL in seconds, which is converted to an absolute UTC expiry timestamp. On retrieval, expired entries are detected by comparing the stored expiry against the current time and are lazily purged from the database. This approach avoids the need for background cleanup processes while ensuring expired data is never returned.
The class supports both synchronous and asynchronous interfaces. The async methods (`aget`, `aset`, `aclear`) delegate to the synchronous implementations via `asyncio.to_thread`, making the cache safe to use in async LangGraph applications without blocking the event loop. Namespaced keys allow logical partitioning of cached data, with the `clear` method supporting selective namespace-based eviction or full cache clearing.
Usage
Use `SqliteCache` when you need a lightweight, zero-dependency cache that persists across process restarts. It is ideal for local development, testing, or single-process deployments where Redis or another external cache is not available. Configure it with a file path and an optional custom serializer.
Code Reference
Source Location
Signature
class SqliteCache(BaseCache[ValueT]):
def __init__(
self,
*,
path: str,
serde: SerializerProtocol | None = None,
) -> None: ...
def get(self, keys: Sequence[FullKey]) -> dict[FullKey, ValueT]: ...
async def aget(self, keys: Sequence[FullKey]) -> dict[FullKey, ValueT]: ...
def set(self, mapping: Mapping[FullKey, tuple[ValueT, int | None]]) -> None: ...
async def aset(self, mapping: Mapping[FullKey, tuple[ValueT, int | None]]) -> None: ...
def clear(self, namespaces: Sequence[Namespace] | None = None) -> None: ...
async def aclear(self, namespaces: Sequence[Namespace] | None = None) -> None: ...
Import
from langgraph.cache.sqlite import SqliteCache
I/O Contract
Constructor Parameters
| Parameter |
Type |
Required |
Description
|
| path |
`str` |
Yes |
File system path for the SQLite database
|
| serde |
None` |
No |
Custom serializer; defaults to the base class default
|
get
| Parameter |
Type |
Description
|
| keys |
`Sequence[FullKey]` |
List of `(Namespace, key_str)` tuples to look up
|
| Return Type |
Description
|
| `dict[FullKey, ValueT]` |
Mapping of found (non-expired) keys to their deserialized values
|
set
| Parameter |
Type |
Description
|
| mapping |
None]]` |
Keys mapped to `(value, ttl_seconds)` pairs; TTL of `None` means no expiry
|
clear
| Parameter |
Type |
Description
|
| namespaces |
None` |
Namespaces to clear; `None` clears all cached values
|
Usage Examples
from langgraph.cache.sqlite import SqliteCache
# Create a file-backed cache
cache = SqliteCache(path="/tmp/langgraph_cache.db")
# Store a value with a 60-second TTL
ns = ("my_app", "embeddings")
cache.set({
(ns, "doc_123"): ({"embedding": [0.1, 0.2, 0.3]}, 60),
})
# Retrieve the cached value
results = cache.get([(ns, "doc_123")])
if (ns, "doc_123") in results:
print(results[(ns, "doc_123")]) # {"embedding": [0.1, 0.2, 0.3]}
# Clear a specific namespace
cache.clear(namespaces=[ns])
# Clear all cached values
cache.clear()
Related Pages