PickleCache: Basic Persistent Local Caching#

class my.caches.PickleCache.PickleCache(*, file: Path, func: Callable[[], ~collections.abc.Coroutine[None, None, dict[Key, Value]]] | None=None, data: dict[Key, Value]={}, ttl: timedelta = datetime.timedelta(days=1), last_read: datetime = <factory>, last_write: datetime = <factory>)#

Persistent cache backed by pickle files with TTL-based invalidation.

Supports three data sources with fallback hierarchy:

  1. In-memory data (fastest, if fresh)

  2. Pickle file on disk (if within TTL)

  3. Async callback function (if provided, refreshes cache)

Data is automatically written to disk when refreshed from callback.

Examples

Persist data to disk, then read it back through a fresh instance:

>>> import asyncio, tempfile
>>> from my import PickleCache
>>> tmp = tempfile.TemporaryDirectory()
>>> cache = PickleCache(file=f'{tmp.name}/data.pkl', data={'a': 1})
>>> cache['b'] = 2
>>> cache.write()
>>> fresh = PickleCache(file=f'{tmp.name}/data.pkl')
>>> asyncio.run(fresh.read())
{'a': 1, 'b': 2}
async PickleCache.read() → dict[Key, Value]#

Refresh cache data, checking memory, disk, and callback in order.

Falls back through data sources:

  1. Returns in-memory data if recent (within TTL)

  2. Loads from pickle file if it exists and is fresh

  3. Calls async func if provided and caches result

Returns:

Dictionary of cached data.

Examples

Populate an empty cache from its async callback:

>>> import asyncio, tempfile
>>> from my import PickleCache
>>> async def fetch():
...     return {'x': 10}
>>> tmp = tempfile.TemporaryDirectory()
>>> cache = PickleCache(file=f'{tmp.name}/cb.pkl', func=fetch)
>>> asyncio.run(cache.read())
{'x': 10}
PickleCache.write() → None#

Write current data to pickle file and update timestamps.

Writes to a temporary file in the same directory and atomically renames it into place, so a crash or kill mid-write can never leave a torn/partially-written pickle file at file.

Warning

This is a trust boundary – only unpickle data you trust. read() calls pickle.loads() on whatever bytes are on disk at file, and unpickling arbitrary/untrusted data can execute arbitrary code. Treat this cache’s file as trusted storage, not a format for exchanging data with other parties.

PickleCache.get(key: Key, default: Any | None = None) → Value | None#

Get value for key, returning default if not found.

PickleCache.items() → Iterable[tuple[Key, Value]]#

Iterate through all key-value pairs in the cache.

PickleCache.keys() → Iterable#

Iterate through all keys in the cache.

PickleCache.values() → Iterable#

Iterate through all values in the cache.