Memoize

memoize

memoize(
    cache: CacheProtocol,
    name: str | None = None,
    *,
    typed: bool = False,
    expire: float | None = None,
    tags: str | Iterable[str] | None = None,
    include: Iterable[str | int] = (),
    exclude: Iterable[str | int] = ()
) -> MemoizedDecorator

Memoizing cache decorator.

Decorator to wrap callable with memoizing function using cache. Repeated calls with the same arguments will lookup result in cache and avoid function evaluation.

If name is set to None (default), the callable name will be determined automatically.

When expire is set to zero, function results will not be set in the cache. Cache lookups still occur, however.

If typed is set to True, function arguments of different types will be cached separately. For example, f(3) and f(3.0) will be treated as distinct calls with distinct results.

The original underlying function is accessible through the __wrapped__ attribute. This is useful for introspection, for bypassing the cache, or for rewrapping the function with a different cache.

from typed_diskcache import Cache
from typed_diskcache.utils.memo import memoize_stampede


cache = Cache()

@cache.memoize(expire=1, tag="fib")
def fibonacci(number):
    if number == 0:
        return 0
    elif number == 1:
        return 1
    else:
        return fibonacci(number - 1) + fibonacci(number - 2)


print(fibonacci(100))
# 354224848179261915075

An additional cache_key method can be used to generate the cache key used for the given arguments.

key = fibonacci.cache_key(100)
print(cache[key])
# 354224848179261915075

Remember to call memoize when decorating a callable. If you forget, then a TypeError will occur. Note the lack of parenthenses after memoize below:

@cache.memoize
def test():
    pass

Parameters:

Name	Type	Description	Default
cache	CacheProtocol	cache to store callable arguments and return values	required
name	str | None	name given for callable (default None, automatic)	None
typed	bool	cache different types separately (default False)	False
expire	float | None	seconds until arguments expire (default None, no expiry)	None
tags	str | Iterable[str] | None	text to associate with arguments (default None)	None
include	Iterable[str | int]	positional or keyword args to include (default ())	()
exclude	Iterable[str | int]	positional or keyword args to exclude (default ())	()

Returns:

Type	Description
MemoizedDecorator	memoized callable decorator

Source code in src/typed_diskcache/utils/memo.py

def memoize(
    cache: CacheProtocol,
    name: str | None = None,
    *,
    typed: bool = False,
    expire: float | None = None,
    tags: str | Iterable[str] | None = None,
    include: Iterable[str | int] = (),
    exclude: Iterable[str | int] = (),
) -> MemoizedDecorator:
    """Memoizing cache decorator.

    Decorator to wrap callable with memoizing function using cache.
    Repeated calls with the same arguments will lookup result in cache and
    avoid function evaluation.

    If name is set to None (default), the callable name will be determined
    automatically.

    When expire is set to zero, function results will not be set in the
    cache. Cache lookups still occur, however.

    If typed is set to True, function arguments of different types will be
    cached separately. For example, f(3) and f(3.0) will be treated as
    distinct calls with distinct results.

    The original underlying function is accessible through the `__wrapped__`
    attribute. This is useful for introspection, for bypassing the cache,
    or for rewrapping the function with a different cache.

    ```python
    from typed_diskcache import Cache
    from typed_diskcache.utils.memo import memoize_stampede


    cache = Cache()

    @cache.memoize(expire=1, tag="fib")
    def fibonacci(number):
        if number == 0:
            return 0
        elif number == 1:
            return 1
        else:
            return fibonacci(number - 1) + fibonacci(number - 2)


    print(fibonacci(100))
    # 354224848179261915075
    ```

    An additional `cache_key` method can be used to generate the
    cache key used for the given arguments.

    ```python
    key = fibonacci.cache_key(100)
    print(cache[key])
    # 354224848179261915075
    ```

    Remember to call memoize when decorating a callable. If you forget,
    then a TypeError will occur. Note the lack of parenthenses after
    memoize below:

    ```python
    @cache.memoize
    def test():
        pass
    ```

    Args:
        cache: cache to store callable arguments and return values
        name: name given for callable (default None, automatic)
        typed: cache different types separately (default False)
        expire: seconds until arguments expire (default None, no expiry)
        tags: text to associate with arguments (default None)
        include: positional or keyword args to include (default ())
        exclude: positional or keyword args to exclude (default ())

    Returns:
        memoized callable decorator
    """
    tags = frozenset(tags or ())
    include = frozenset(include)
    exclude = frozenset(exclude)

    def decorator(func: Callable[..., Any]) -> Memoized[..., Any]:
        base = name or full_name(func)
        memoized = AsyncMemoized if inspect.iscoroutinefunction(func) else Memoized
        return memoized(
            cache=cache,
            func=func,
            base=base,
            typed=typed,
            expire=expire,
            tags=tags,
            include=include,
            exclude=exclude,
        )

    return decorator  # pyright: ignore[reportReturnType]

memoize_stampede

memoize_stampede(
    cache: CacheProtocol,
    name: str | None = None,
    *,
    typed: bool = False,
    expire: float | None = None,
    tags: str | Iterable[str] | None = None,
    include: Iterable[str | int] = (),
    exclude: Iterable[str | int] = (),
    beta: float = 1
) -> MemoizedStampedeDecorator

Memoizing cache decorator with cache stampede protection.

Cache stampedes are a type of system overload that can occur when parallel computing systems using memoization come under heavy load. This behaviour is sometimes also called dog-piling, cache miss storm, cache choking, or the thundering herd problem.

The memoization decorator implements cache stampede protection through early recomputation. Early recomputation of function results will occur probabilistically before expiration in a background thread of execution. Early probabilistic recomputation is based on research by Vattani, A.; Chierichetti, F.; Lowenstein, K. (2015), Optimal Probabilistic Cache Stampede Prevention, VLDB, pp. 886-897, ISSN 2150-8097

If name is set to None (default), the callable name will be determined automatically.

If typed is set to True, function arguments of different types will be cached separately. For example, f(3) and f(3.0) will be treated as distinct calls with distinct results.

The original underlying function is accessible through the __wrapped__ attribute. This is useful for introspection, for bypassing the cache, or for rewrapping the function with a different cache.

from typed_diskcache import Cache
from typed_diskcache.utils.memo import memoize_stampede


cache = Cache()

@memoize_stampede(cache, expire=1)
def fib(number):
    if number == 0:
        return 0
    elif number == 1:
        return 1
    else:
        return fib(number - 1) + fib(number - 2)


print(fib(100))
# 354224848179261915075

An additional cache_key method can be used to generate the cache key used for the given arguments.

key = fib.cache_key(100)
del cache[key]

Remember to call memoize when decorating a callable. If you forget, then a TypeError will occur.

Parameters:

Name	Type	Description	Default
cache	CacheProtocol	cache to store callable arguments and return values	required
name	str | None	name given for callable (default None, automatic)	None
typed	bool	cache different types separately (default False)	False
expire	float | None	seconds until arguments expire	None
tags	str | Iterable[str] | None	text to associate with arguments (default None)	None
include	Iterable[str | int]	positional or keyword args to include (default ())	()
exclude	Iterable[str | int]	positional or keyword args to exclude (default ())	()
beta	float	cache stampede protection factor (default 1)	1

Returns:

Type	Description
MemoizedStampedeDecorator	memoized callable decorator

Source code in src/typed_diskcache/utils/memo.py

def memoize_stampede(
    cache: CacheProtocol,
    name: str | None = None,
    *,
    typed: bool = False,
    expire: float | None = None,
    tags: str | Iterable[str] | None = None,
    include: Iterable[str | int] = (),
    exclude: Iterable[str | int] = (),
    beta: float = 1,
) -> MemoizedStampedeDecorator:
    """Memoizing cache decorator with cache stampede protection.

    Cache stampedes are a type of system overload that can occur when parallel
    computing systems using memoization come under heavy load. This behaviour
    is sometimes also called dog-piling, cache miss storm, cache choking, or
    the thundering herd problem.

    The memoization decorator implements cache stampede protection through
    early recomputation. Early recomputation of function results will occur
    probabilistically before expiration in a background thread of
    execution. Early probabilistic recomputation is based on research by
    Vattani, A.; Chierichetti, F.; Lowenstein, K. (2015), Optimal Probabilistic
    Cache Stampede Prevention, VLDB, pp. 886-897, ISSN 2150-8097

    If name is set to None (default), the callable name will be determined
    automatically.

    If typed is set to True, function arguments of different types will be
    cached separately. For example, f(3) and f(3.0) will be treated as distinct
    calls with distinct results.

    The original underlying function is accessible through the `__wrapped__`
    attribute. This is useful for introspection, for bypassing the cache, or
    for rewrapping the function with a different cache.

    ```python
    from typed_diskcache import Cache
    from typed_diskcache.utils.memo import memoize_stampede


    cache = Cache()

    @memoize_stampede(cache, expire=1)
    def fib(number):
        if number == 0:
            return 0
        elif number == 1:
            return 1
        else:
            return fib(number - 1) + fib(number - 2)


    print(fib(100))
    # 354224848179261915075
    ```

    An additional `cache_key` method can be used to generate the cache
    key used for the given arguments.

    ```python
    key = fib.cache_key(100)
    del cache[key]
    ```

    Remember to call memoize when decorating a callable. If you forget, then a
    TypeError will occur.

    Args:
        cache: cache to store callable arguments and return values
        name: name given for callable (default None, automatic)
        typed: cache different types separately (default False)
        expire: seconds until arguments expire
        tags: text to associate with arguments (default None)
        include: positional or keyword args to include (default ())
        exclude: positional or keyword args to exclude (default ())
        beta: cache stampede protection factor (default 1)

    Returns:
        memoized callable decorator
    """
    tags = frozenset(tags or ())
    include = frozenset(include)
    exclude = frozenset(exclude)

    def decorator(func: Callable[..., Any]) -> MemoizedStampede[..., Any]:
        base = name or full_name(func)
        memoized = (
            AsyncMemoizedStampede
            if inspect.iscoroutinefunction(func)
            else MemoizedStampede
        )
        return memoized(
            cache=cache,
            func=func,
            base=base,
            typed=typed,
            expire=expire,
            tags=tags,
            include=include,
            exclude=exclude,
            beta=beta,
        )

    return decorator  # pyright: ignore[reportReturnType]

MemoizedDecorator

Bases: Protocol

__call__

__call__(
    func: Callable[_P, Coroutine[Any, Any, _T]],
) -> AsyncMemoized[_P, _T]

__call__(func: Callable[_P, _T]) -> Memoized[_P, _T]

__call__(func: Callable[_P, Any]) -> Memoized[_P, Any]

__call__(func: Callable[_P, Any]) -> Memoized[_P, Any]

Source code in src/typed_diskcache/utils/memo.py

def __call__(self, func: Callable[_P, Any]) -> Memoized[_P, Any]: ...

MemoizedStampedeDecorator

Bases: Protocol

__call__

__call__(
    func: Callable[_P, Coroutine[Any, Any, _T]],
) -> AsyncMemoizedStampede[_P, _T]

__call__(
    func: Callable[_P, _T],
) -> MemoizedStampede[_P, _T]

__call__(
    func: Callable[_P, Any],
) -> MemoizedStampede[_P, Any]

__call__(
    func: Callable[_P, Any],
) -> MemoizedStampede[_P, Any]

Source code in src/typed_diskcache/utils/memo.py

def __call__(self, func: Callable[_P, Any]) -> MemoizedStampede[_P, Any]: ...

AsyncMemoized

AsyncMemoized(
    *,
    cache: CacheProtocol,
    func: Callable[_P, Coroutine[Any, Any, _T]],
    base: str,
    typed: bool,
    expire: float | None,
    tags: frozenset[str],
    include: frozenset[str | int],
    exclude: frozenset[str | int]
)

Bases: Memoized[_P, Coroutine[Any, Any, _T]], Generic[_P, _T]

Source code in src/typed_diskcache/utils/memo.py

@override
def __init__(
    self,
    *,
    cache: CacheProtocol,
    func: Callable[_P, Coroutine[Any, Any, _T]],
    base: str,
    typed: bool,
    expire: float | None,
    tags: frozenset[str],
    include: frozenset[str | int],
    exclude: frozenset[str | int],
) -> None:
    super().__init__(
        cache=cache,
        func=func,
        base=base,
        typed=typed,
        expire=expire,
        tags=tags,
        include=include,
        exclude=exclude,
    )

__wrapped__ property

__wrapped__: Callable[_P, Coroutine[Any, Any, _T]]

__call__ async

__call__(*args: args, **kwargs: kwargs) -> _T

Source code in src/typed_diskcache/utils/memo.py

@override
async def __call__(self, *args: _P.args, **kwargs: _P.kwargs) -> _T:
    key = self.cache_key(*args, **kwargs)
    container = await self._cache.aget(key, retry=True)
    if not container.default:
        return container.value

    value = await self._func(*args, **kwargs)
    if self._expire is None or self._expire > 0:
        await self._cache.aset(
            key, value, expire=self._expire, tags=self._tags, retry=True
        )
    return value

cache_key

cache_key(*args: args, **kwargs: kwargs) -> tuple[Any, ...]

Source code in src/typed_diskcache/utils/memo.py

def cache_key(self, *args: _P.args, **kwargs: _P.kwargs) -> tuple[Any, ...]:
    return args_to_key(
        base=self._base,
        args=args,
        kwargs=kwargs,
        typed=self._typed,
        include=self._include,
        exclude=self._exclude,
    )

Memoized

Memoized(
    *,
    cache: CacheProtocol,
    func: Callable[_P, _T],
    base: str,
    typed: bool,
    expire: float | None,
    tags: frozenset[str],
    include: frozenset[str | int],
    exclude: frozenset[str | int]
)

Bases: Generic[_P, _T]

Source code in src/typed_diskcache/utils/memo.py

def __init__(
    self,
    *,
    cache: CacheProtocol,
    func: Callable[_P, _T],
    base: str,
    typed: bool,
    expire: float | None,
    tags: frozenset[str],
    include: frozenset[str | int],
    exclude: frozenset[str | int],
) -> None:
    self._cache = cache
    self._func = func
    self._base = base
    self._typed = typed
    self._expire = expire
    self._tags = tags
    self._include = include
    self._exclude = exclude

__wrapped__ property

__wrapped__: Callable[_P, _T]

__call__

__call__(*args: args, **kwargs: kwargs) -> _T

Source code in src/typed_diskcache/utils/memo.py

def __call__(self, *args: _P.args, **kwargs: _P.kwargs) -> _T:
    key = self.cache_key(*args, **kwargs)
    container = self._cache.get(key, retry=True)
    if not container.default:
        return container.value

    value = self._func(*args, **kwargs)
    if self._expire is None or self._expire > 0:
        self._cache.set(
            key, value, expire=self._expire, tags=self._tags, retry=True
        )
    return value

cache_key

cache_key(*args: args, **kwargs: kwargs) -> tuple[Any, ...]

Source code in src/typed_diskcache/utils/memo.py

def cache_key(self, *args: _P.args, **kwargs: _P.kwargs) -> tuple[Any, ...]:
    return args_to_key(
        base=self._base,
        args=args,
        kwargs=kwargs,
        typed=self._typed,
        include=self._include,
        exclude=self._exclude,
    )