pydvl.utils.caching.memory

InMemoryCacheBackend

InMemoryCacheBackend()

Bases: CacheBackend

In-memory cache backend that stores results in a dictionary.

Implements the CacheBackend interface for an in-memory-based cache. Stores cache entries as values in a dictionary, keyed by cache key. This allows sharing evaluations across threads in a single process.

The implementation is not thread-safe.

ATTRIBUTE	DESCRIPTION
cached_values	Dictionary used to store cached values. TYPE: Dict[str, Any]

Example

Basic usage:

>>> from pydvl.utils.caching.memory import InMemoryCacheBackend
>>> cache_backend = InMemoryCacheBackend()
>>> cache_backend.clear()
>>> value = 42
>>> cache_backend.set("key", value)
>>> cache_backend.get("key")
42

Callable wrapping:

>>> from pydvl.utils.caching.memory import InMemoryCacheBackend
>>> cache_backend = InMemoryCacheBackend()
>>> cache_backend.clear()
>>> value = 42
>>> def foo(x: int):
...     return x + 1
...
>>> wrapped_foo = cache_backend.wrap(foo)
>>> wrapped_foo(value)
43
>>> wrapped_foo.stats.misses
1
>>> wrapped_foo.stats.hits
0
>>> wrapped_foo(value)
43
>>> wrapped_foo.stats.misses
1
>>> wrapped_foo.stats.hits
1

Source code in src/pydvl/utils/caching/memory.py

def __init__(self) -> None:
    """Initialize the in-memory cache backend."""
    super().__init__()
    self.cached_values: Dict[str, Any] = {}

wrap

wrap(
    func: Callable, *, config: Optional[CachedFuncConfig] = None
) -> CachedFunc

Wraps a function to cache its results.

PARAMETER	DESCRIPTION
func	The function to wrap. TYPE: Callable
config	Optional caching options for the wrapped function. TYPE: Optional[CachedFuncConfig] DEFAULT: None

RETURNS	DESCRIPTION
CachedFunc	The wrapped cached function.

Source code in src/pydvl/utils/caching/base.py

def wrap(
    self,
    func: Callable,
    *,
    config: Optional[CachedFuncConfig] = None,
) -> "CachedFunc":
    """Wraps a function to cache its results.

    Args:
        func: The function to wrap.
        config: Optional caching options for the wrapped function.

    Returns:
        The wrapped cached function.
    """
    return CachedFunc(
        func,
        cache_backend=self,
        config=config,
    )

get

get(key: str) -> Optional[Any]

Get a value from the cache.

PARAMETER	DESCRIPTION
key	Cache key. TYPE: str

RETURNS	DESCRIPTION
Optional[Any]	Cached value or None if not found.

Source code in src/pydvl/utils/caching/memory.py

def get(self, key: str) -> Optional[Any]:
    """Get a value from the cache.

    Args:
        key: Cache key.

    Returns:
        Cached value or None if not found.
    """
    value = self.cached_values.get(key, None)
    if value is not None:
        self.stats.hits += 1
    else:
        self.stats.misses += 1
    return value

set

set(key: str, value: Any) -> None

Set a value in the cache.

PARAMETER	DESCRIPTION
key	Cache key. TYPE: str
value	Value to cache. TYPE: Any

Source code in src/pydvl/utils/caching/memory.py

def set(self, key: str, value: Any) -> None:
    """Set a value in the cache.

    Args:
        key: Cache key.
        value: Value to cache.
    """
    self.cached_values[key] = value
    self.stats.sets += 1

clear

clear() -> None

Deletes cache dictionary and recreates it.

Source code in src/pydvl/utils/caching/memory.py

def clear(self) -> None:
    """Deletes cache dictionary and recreates it."""
    del self.cached_values
    self.cached_values = {}

combine_hashes

combine_hashes(*args: str) -> str

Join cache key components.

Source code in src/pydvl/utils/caching/memory.py

def combine_hashes(self, *args: str) -> str:
    """Join cache key components."""
    return os.pathsep.join(args)