pydvl.utils.caching.base

CacheStats dataclass

CacheStats(
    sets: int = 0,
    misses: int = 0,
    hits: int = 0,
    timeouts: int = 0,
    errors: int = 0,
    reconnects: int = 0,
)

Class used to store statistics gathered by cached functions.

ATTRIBUTE	DESCRIPTION
sets	Number of times a value was set in the cache. TYPE: int
misses	Number of times a value was not found in the cache. TYPE: int
hits	Number of times a value was found in the cache. TYPE: int
timeouts	Number of times a timeout occurred. TYPE: int
errors	Number of times an error occurred. TYPE: int
reconnects	Number of times the client reconnected to the server. TYPE: int

CacheResult dataclass

CacheResult(value: float, count: int = 1, variance: float = 0.0)

A class used to store the cached result of a computation as well as count and variance when using repeated evaluation.

ATTRIBUTE	DESCRIPTION
value	Cached value. TYPE: float
count	Number of times this value has been computed. TYPE: int
variance	Variance associated with the cached value. TYPE: float

CacheBackend

CacheBackend()

Bases: ABC

Abstract base class for cache backends.

Defines interface for cache access including wrapping callables, getting/setting results, clearing cache, and combining cache keys.

ATTRIBUTE	DESCRIPTION
stats	Cache statistics tracker.

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

def __init__(self) -> None:
    self.stats = CacheStats()

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 abstractmethod

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

Abstract method to retrieve a cached result.

Implemented by subclasses.

PARAMETER	DESCRIPTION
key	The cache key. TYPE: str

RETURNS	DESCRIPTION
Optional[CacheResult]	The cached result or None if not found.

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

@abstractmethod
def get(self, key: str) -> Optional[CacheResult]:
    """Abstract method to retrieve a cached result.

    Implemented by subclasses.

    Args:
        key: The cache key.

    Returns:
        The cached result or None if not found.
    """
    pass

set abstractmethod

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

Abstract method to set a cached result.

Implemented by subclasses.

PARAMETER	DESCRIPTION
key	The cache key. TYPE: str
value	The result to cache. TYPE: CacheResult

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

@abstractmethod
def set(self, key: str, value: CacheResult) -> None:
    """Abstract method to set a cached result.

    Implemented by subclasses.

    Args:
        key: The cache key.
        value: The result to cache.
    """
    pass

clear abstractmethod

clear() -> None

Abstract method to clear the entire cache.

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

@abstractmethod
def clear(self) -> None:
    """Abstract method to clear the entire cache."""
    pass

combine_hashes abstractmethod

combine_hashes(*args: str) -> str

Abstract method to combine cache keys.

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

@abstractmethod
def combine_hashes(self, *args: str) -> str:
    """Abstract method to combine cache keys."""
    pass

CachedFunc

CachedFunc(
    func: Callable[..., float],
    *,
    cache_backend: CacheBackend,
    config: Optional[CachedFuncConfig] = None
)

Caches callable function results with a provided cache backend.

Wraps a callable function to cache its results using a provided an instance of a subclass of CacheBackend.

This class is heavily inspired from that of joblib.memory.MemorizedFunc.

This class caches calls to the wrapped callable by generating a hash key based on the wrapped callable's code, the arguments passed to it and the optional hash_prefix.

Warning

This class only works with hashable arguments to the wrapped callable.

PARAMETER	DESCRIPTION
func	Callable to wrap. TYPE: Callable[..., float]
cache_backend	Instance of CacheBackendBase that handles setting and getting values. TYPE: CacheBackend
config	Configuration for wrapped function. TYPE: Optional[CachedFuncConfig] DEFAULT: None

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

def __init__(
    self,
    func: Callable[..., float],
    *,
    cache_backend: CacheBackend,
    config: Optional[CachedFuncConfig] = None,
) -> None:
    self.func = func
    self.cache_backend = cache_backend
    if config is None:
        config = CachedFuncConfig()
    self.config = config

    self.__doc__ = f"A wrapper around {func.__name__}() with caching enabled.\n" + (
        CachedFunc.__doc__ or ""
    )
    self.__name__ = f"cached_{func.__name__}"
    path = list(reversed(func.__qualname__.split(".")))
    patched = [f"cached_{path[0]}"] + path[1:]
    self.__qualname__ = ".".join(reversed(patched))

stats property

stats: CacheStats

Cache backend statistics.

__call__

__call__(*args, **kwargs) -> float

Call the wrapped cached function.

Executes the wrapped function, caching and returning the result.

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

def __call__(self, *args, **kwargs) -> float:
    """Call the wrapped cached function.

    Executes the wrapped function, caching and returning the result.
    """
    return self._cached_call(args, kwargs)