File System Cache Decorator in Python

fscache.py

""" Caches expensive function calls in pickled bytes on disk. """

import os
import shutil
import subprocess
import dill
from functools import wraps
import hashlib
import base64


def clear_caches():
    """ Delete all cache directories created by fscache """
    for dirname in filter(lambda s: s.startswith('.fscache'), os.listdir('./')):
        shutil.rmtree(dirname)


def clear_cache(context: str):
    """ Delete cache directory for this context """
    try:
        shutil.rmtree(_dirname_for(context))
    except FileNotFoundError:
        pass


def get(context: str, key: str):
    """ Get object from cache if present, return None otherwise """
    directory = _dirname_for(context)
    if directory not in os.listdir('./'):
        subprocess.call(['mkdir', '-p', directory])
    if key in os.listdir(directory + '/'):
        with open('{}/{}'.format(directory, key), 'rb') as infile:
            return dill.load(infile)


def put(context: str, key: str, obj, strict_exceptions=False):
    """ Put object to file system cache, maybe die on invalid key name """
    directory = _dirname_for(context)
    if directory in os.listdir('./'):
        try:
            with open('{}/{}'.format(directory, key), 'wb') as outfile:
                    dill.dump(obj, outfile)
        except Exception as e:
            if strict_exceptions:
                raise e
            else:
                print(e)


def _dirname_for(context: str):
    return '.fscache-{}'.format(context)


def default_key_fn(*args, **kwargs) -> str:
    kw_pairs = tuple('{}={}'.format(k, v) for k, v in kwargs.items())
    hasher = hashlib.md5()
    hasher.update(':'.join(map(str, args + kw_pairs)).encode('utf-8'))
    return base64.standard_b64encode(hasher.digest()).decode('utf-8').replace('/', '+') + '.dill'


def caches(context: str, key_fn=default_key_fn):
    def outer_wrapper(fn):
        @wraps(fn)
        def wrapper(*args, **kwargs):
            key = key_fn(*args, **kwargs)
            maybe = get(context, key)
            if maybe is not None:
                return maybe
            result = fn(*args, **kwargs)
            put(context, key, result)
            return result
        return wrapper
    return outer_wrapper


# Invoking function creates a .example directory, caching increment results in number named
# files, like `.fscache-example/12.dill`, which contains a pickled Python int 13
@caches('example')
def inc(num):
    return num + 1