`duck.utils.multiprocessing`¶

ProcessSafeLRUCache: A process-safe, shared-memory LRU cache for Python multi-process applications.

This class allows multiple processes to share and update a central in-memory cache, with Least Recently Used (LRU) eviction and optional per-key expiry, using Python’s multiprocessing.Manager for process safety.

Features:

Process-safe via Manager-proxied dicts and lists.
True LRU eviction: oldest unused entry is removed once maxkeys is exceeded.
Optional per-key expiry: keys are removed after a set time.
API is similar to standard in-memory LRU cache: get, set, delete, has, clear.
Locking ensures atomicity for compound operations.
Suitable for multi-process WSGI/ASGI, background jobs, ML/AI workers, caching across service forks.

Example use cases:

Share a cache between worker processes in Gunicorn, Uvicorn, or multiprocessing scripts.
Protect frequently-accessed data with expiry and auto-eviction.
Use in ML/AI pipeline for process-wide memoization or inference result caching.

Usage Example:

import time
import multiprocessing

from duck.utils.multiprocesing import ProcessSafeLRUCache

# Must be called in __main__ for Windows fork support!
if __name__ == "__main__":
    manager = multiprocessing.Manager()
    cache = ProcessSafeLRUCache(maxkeys=5)

    def store_and_read(cache, key, value):
        cache.set(key, value, expiry=10)
        print("Stored", key, "got back", cache.get(key))

    ps = []
    for i in range(10):
        p = multiprocessing.Process(target=store_and_read, args=(cache, str(i), i * 2))
        p.start()
        ps.append(p)
        time.sleep(1)

    for p in ps:
        p.join()

    print("Final shared cache:", dict(cache.cache))

Submodules¶

Package Contents¶

Classes¶

ProcessSafeLRUCache

ProcessSafeLRUCache(maxkeys=None)

API¶

class duck.utils.multiprocessing.ProcessSafeLRUCache(maxkeys=None)[source]¶

ProcessSafeLRUCache(maxkeys=None)

A process-safe, shared-memory LRU cache with per-key expiry time.

Uses multiprocessing.Manager for process-safe shared dict/list storage.

Parameters:: maxkeys – Optional maximum number of cache entries (evicts oldest if exceeded).

Methods:

set(key, value, expiry=None): Set key to value, optionally with expiry time in seconds.
get(key, default=None, pop=False): Get value for key (None/default if missing), optionally popping it.
has(key): Check if key exists (not evicted/expired).
delete(key): Delete a cache entry.
clear(): Remove all cache and expiry data.
close(): Aliased to clear.

Initialization

__slots__¶: None

clear()[source]¶: Clear the entire cache and expiry info.

close()[source]¶: Alias for clear().

delete(key: str)[source]¶: Remove a key from the cache.

get(key: str, default=None, pop=False)[source]¶

Get a value from the cache (or default if missing/expired).

Parameters:

key – Cache key.
default – Value to return if key not present.
pop – If True, remove the key after retrieval.

Returns:

Cached value, or default/None.

has(key: str)[source]¶: Check if the cache has a key (not expired/evicted).

set(key: str, value, expiry=None)[source]¶

Set a value in the cache, marking it as most recently used.