Skip to content

Instantly share code, notes, and snippets.

@malderete
Created July 5, 2016 15:02
Show Gist options
  • Save malderete/a0467603085be1e31971b6eeb0b5fdb3 to your computer and use it in GitHub Desktop.
Save malderete/a0467603085be1e31971b6eeb0b5fdb3 to your computer and use it in GitHub Desktop.
Time based cache decorator
import functools
import logging
import time
logger = logging.getLogger(__name__)
MAX_ENTRIES = 100 # Max items per cache!
def memoize(expire_seconds):
"""Time-Expiration based cache decorator.
Arguments to the cached function must be hashable.
Clear the cache using f.cache_clear().
Debug the interval cache using f.get_cache().
"""
placeholder = object()
def decorating(func):
cache = {}
@functools.wraps(func)
def wrapper(*args, **kwargs):
key = functools._make_key(args, kwargs, False)
cache_info = cache.get(key, placeholder)
if cache_info is not placeholder:
# Cache Hit!
now = time.monotonic()
if now <= cache_info[0]:
# Cache still valid
logger.debug('Cache found %s for %s', key, func.__name__)
return cache_info[1]
else:
# Try to make room cleaning this key
logger.debug('Cache expired %s for %s', key, func.__name__)
try:
del cache[key]
except KeyError:
# Other thread could have deleted the key
pass
result = func(*args, **kwargs)
# If we have room save the result
if len(cache) < MAX_ENTRIES:
logger.debug('Cache saving key %s for %s result %s', key, func.__name__, result)
expiration = time.monotonic() + expire_seconds
cache[key] = (expiration, result)
return result
def clear_cache():
"""Clean the interval cache of the user function."""
cache.clear()
def get_cache():
"""Return the interval cache (useful for debugging)."""
return cache
# Public API to manipulate the cache from outside
wrapper.clear_cache = clear_cache
wrapper.get_cache = get_cache
return wrapper
return decorating
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment