Source code for miniutils.caching.file_call

import os
import shelve
from collections import namedtuple
from functools import wraps
from glob import glob

from miniutils.opt_decorator import optional_argument_decorator
from miniutils.logs_base import debug


[docs]class FileCached: def __init__(self, fn, cache_path=None, files_used=None, auto_purge=False): """Caches function results to a file to save re-computation of highly expensive calls :param fn: The functions whose result should be cached :type fn: function :param cache_path: No-extension file path where cache should be kept :type cache_path: str :param files_used: List of files that could effect the result of this function; cache results are invalidated if any of these files are updated since the last function call :type files_used: Iterable :param auto_purge: If True, deletes the file cache when this cache object passes out of scope :type: auto_purge: bool """ self.__wrapped__ = fn self.path = cache_path or '.__cache_{}'.format(fn.__name__) self.files_used = tuple(sorted([os.path.abspath(os.path.expanduser(p)) for p in (files_used or [])])) self._shelf = shelve.open(self.path) self._auto_purge = auto_purge self._hits = 0 self._misses = 0 def __call__(self, *args, **kwargs): key = ':'.join(self.files_used) + ':' + hex(hash((args, tuple(sorted(kwargs.items()))))) if key in self._shelf: file_update_times, result = self._shelf[key] for file in self.files_used: if not os.path.exists(file) or os.path.getmtime(file) > file_update_times[file]: break else: self._hits += 1 return result self._misses += 1 file_update_times = {file: os.path.getmtime(file) for file in self.files_used} result = self.__wrapped__(*args, **kwargs) self._shelf[key] = (file_update_times, result) return result def __del__(self): if self._auto_purge: self.cache_clear(create_new_shelf=False)
[docs] def cache_clear(self, create_new_shelf=True): """Deletes the underlying cache""" # TODO: Remove these debug loops debug("Clearing shelf: directory starts with the following files:") for path in glob(os.path.dirname(self.path)): debug(path) del self._shelf for path in glob(self.path + '*'): os.remove(path) debug("Clearing shelf: directory ends with the following files:") for path in glob(os.path.dirname(self.path)): debug(path) if create_new_shelf: self._shelf = shelve.open(self.path)
[docs] def cache_info(self): """Gets information about this cache. :return: A named tuple containing the number of cache ``hits`` and ``misses`` """ return namedtuple('CacheInfo', ('hits', 'misses'))(self._hits, self._misses)
[docs]@optional_argument_decorator def file_cached_decorator(*args, **kwargs): """A decorator version of ``FileCached`` :param cache_path: No-extension file path where cache should be kept :type cache_path: str :param files_used: List of files that could effect the result of this function; cache results are invalidated if any of these files are updated since the last function call :type files_used: Iterable :param auto_purge: If True, deletes the file cache when this cache object passes out of scope :type: auto_purge: bool :return: A decorator for a function :rtype: function """ @wraps(FileCached.__init__) def decorator(fn): """ :rtype: FileCached """ return FileCached(fn, *args, **kwargs) return decorator