skippable_cache.py

from functools import wraps, lru_cache
from contextlib import contextmanager

class _SkipCache(Exception):
    def __init__(self, value):
        self.value = value

class _CachedFunction(object):
    def __init__(self, maxsize = 128, typed = False):
        self.__is_skipped = False
        self.__skip_value = None

        @lru_cache(maxsize, typed)
        def cache_results(f, *args, **kwargs):
            self.__is_skipped = False
            result = f(*args, **kwargs)
            if self.__is_skipped:
                raise _SkipCache(self.__skip_value)
            return result
        self.__cache_results = cache_results

    def __cache_skip(self, value):
        self.__is_skipped = True
        self.__skip_value = value

    def __call__(self, func):
        @wraps(func)
        def _cache_wrapper(*args, **kwargs):
            try:
                return self.__cache_results(func, *args, **kwargs)
            except _SkipCache as skipped:
                return skipped.value

        setattr(_cache_wrapper, 'cache_skip', self.__cache_skip)
        setattr(_cache_wrapper, 'cache_info', self.__cache_results.cache_info)
        setattr(_cache_wrapper, 'cache_clear', self.__cache_results.cache_clear)
        return _cache_wrapper

def cache(maxsize = 128, typed = False):
    """Extend `functools.lru_cache` with a `cache_skip` method to avoid caching "bad" results.

    >>> @cache()
    ... def foo(a):
    ...   if 10 < a:
    ...     return foo.cache_skip(42)
    ...   return 2 * a
    ...

    It works just like regular `lru_cache`:
    >>> [foo(i) for i in range(3)]
    [0, 2, 4]
    >>> [foo(2) for i in range(10)]
    [4, 4, 4, 4, 4, 4, 4, 4, 4, 4]
    >>> foo.cache_info()
    CacheInfo(hits=10, misses=3, maxsize=128, currsize=3)
    >>> foo.cache_clear()
    >>> foo.cache_info()
    CacheInfo(hits=0, misses=0, maxsize=128, currsize=0)

    But instruments the function with a `cache_skip` function to avoid caching
    "unwanted" values:
    >>> prev = foo.cache_info()
    >>> {foo(i) for i in range(11, 21)}
    {42}
    >>> foo.cache_info() == prev
    True
    """
    return _CachedFunction(maxsize, typed)

A wrapper around functools.lru_cache that allows skipping the cache for "unwanted" values.
For example, a function retrieves a user's ID from the DB, but might return None if it can't connect. We don't want to "fix" the None value for that user, so we skip the cache. Next time we'll call the function it'll attempt connecting again.