FileCache: Optimized Persistent Local Caching#

class my.caches.FileCache.FileCache(directory: Annotated[Path, PathType(path_type=dir)], writer: Writer | None = None, reader: Reader | None = None, splitter: Splitter | None = None, max_size: int = 0)#

Two-level file-backed cache with in-memory LRU and on-disk persistence.

Organizes items into a directory structure: group/prefix/file where prefix is derived from file. Maintains separate indices for in-memory (hot) items and on-disk (cold) files. Automatically prunes memory cache and writes to disk when size limits are exceeded:

  • items: In-memory cache of deserialized data

  • files: Index of on-disk files with their contained item names

Examples

Write an item, read it back, then shunt its shard between disk and memory:

>>> import tempfile
>>> from pathlib import Path
>>> from my import FileCache
>>> tmp = tempfile.TemporaryDirectory()
>>> cache = FileCache(Path(tmp.name))
>>> cache.write('users', 'robb_doering', {'role': 'author'})
>>> cache[('users', 'robb_doering')]
{'role': 'author'}
>>> cache.move_to_sys('users', 'robb')  # written to <dir>/users/r/o/b/robb.json
{'robb_doering'}
>>> cache.move_to_mem('users', 'robb')
{'robb_doering': {'role': 'author'}}

I Primary Methods#

FileCache.prune(n: int = 0)#

Write oldest items to disk and remove from memory cache.

Proportionally prunes from each group based on its size relative to total. Items are written to disk before removal from memory.

Parameters:

n – Number of items to prune (default: half of max_size).

FileCache.group_isize(group: str) → int#

Get the number of items currently cached in memory for a group.

FileCache.group_fsize(group: str) → int#

Get the number of items currently cached on disk for a group.

FileCache.read_from_cache(group: str, file: str, prefix: str = '') → dict[str, T] | None#

Read all items from a file, loading from disk if needed.

Parameters:

  • group – Category/namespace.

  • file – File identifier.

  • prefix – Optional prefix override (auto-derived if empty).

Returns:

Dictionary of items, or None if file doesn’t exist.

FileCache.write_to_cache(group: str, file: str, data: dict[str, T], overwrite: bool = True) → None#

Add or replace items in the cache.

Parameters:

  • group – Category/namespace for the items.

  • file – File identifier (prefix derived automatically).

  • data – Dictionary of items to cache.

  • overwrite – Whether to overwrite existing items, or add to them.

FileCache.move_to_sys(group: str, file: str, prefix: str = '') → set[str]#

Write a file’s items directly to disk, overwriting any existing data.

Combines data with any existing cached items for this file before writing. Removes the file from memory cache and adds to disk index.

Parameters:

  • group – Category/namespace.

  • file – File identifier.

  • prefix – Optional prefix override (auto-derived if empty).

Returns:

The set of item names written to disk.

FileCache.move_to_mem(group: str, file: str, prefix: str = '') → dict[str, T]#

Load a file’s items from disk into memory cache.

II Main Interface#

FileCache.read(group: str, name: str) → T | None#

Read a single item by name into memory if it exists, else return None.

Parameters:

  • group – Category/namespace.

  • name – Item identifier (file derived via splitter).

Returns:

The item, or None if not found.

FileCache.write(group: str, name: str, item: T) → None#

Set the value of a single item by name (in memory, for now).

Parameters:

  • group – Category/namespace.

  • name – Item identifier (file derived via splitter).

  • item – Item to store.

FileCache.delete(group: str, name: str) → bool#

Delete a single item by name from the cache and disk.

Removes the item from its shard file, unlinking the file when it becomes empty. Returns False when the item is not found, True when deleted.

Parameters:

  • group – Category/namespace.

  • name – Item identifier (file derived via splitter).

Returns:

Whether the item was found and deleted.

FileCache.flush() → None#

Persist all in-memory shards and clear the memory cache.

Updates the disk index and size counters so flushed items remain readable through this instance. Calling flush on an empty cache is a silent no-op.

FileCache.search(group: str, file_rgx: Pattern, name_rgx: Pattern, prefix: str = '', mode: 'items' | 'files' | 'both' = 'both') → Iterator#

Search for items by file or name patterns.

Parameters:

  • group – Category/namespace to search.

  • file_rgx – Pattern to match against filenames.

  • name_rgx – Pattern to match against item names.

  • prefix – Optional prefix to limit search scope.

  • mode – Search ‘items’ (memory), ‘files’ (disk), or ‘both’.

Yields:

Items matching the search criteria.

Examples

Search across memory and disk:

>>> import regex as re
>>> import tempfile
>>> from pathlib import Path
>>> from my import FileCache
>>> tmp = tempfile.TemporaryDirectory()
>>> cache = FileCache(Path(tmp.name))
>>> cache.write('users', 'robb_doering', {'role': 'author'})
>>> list(cache.search('users', re.compile('robb'), re.compile('robb')))
[{'role': 'author'}]