New cool stuff!

qutil/caching.py

+"""Programming tools"""
+import shelve
+from typing import Callable, Generator, Any
+import inspect
+import tempfile
+import os.path
+import itertools
+import functools
+
+import numpy as np
+
+from qutil.itertools import separate_iterator
+
+__all__ = ["file_cache", "lru_cache"]
+
+
+def to_key(obj: Any) -> Generator[str]:
+    """Convert to a string representation that is unique except for
+    - lists and ndarrays are treated the same
+
+    :param obj:
+    :return:
+    """
+    if isinstance(obj, tuple):
+        element_key_generator = itertools.chain.from_iterable(map(to_key, obj))
+        yield '('
+        yield from separate_iterator(element_key_generator, ',')
+        yield ',)'
+    elif isinstance(obj, (list, np.ndarray)):
+        element_key_generator = itertools.chain.from_iterable(map(to_key, obj))
+        yield '['
+        yield from separate_iterator(element_key_generator, ',')
+        yield ']'
+    elif isinstance(obj, dict):
+        # we need to sort the representations in the case the objects are not sortable
+        sorted_items = sorted(''.join(itertools.chain(to_key(key), (':',), to_key(value)))
+                              for key, value in obj.items())
+        yield '{'
+        yield from separate_iterator(sorted_items, ',')
+        yield '}'
+    elif isinstance(obj, (set, frozenset)):
+        sorted_elements = sorted(map(''.join, map(to_key, obj)))
+        yield '{'
+        yield from separate_iterator(sorted_elements, ',')
+        yield '}'
+    elif isinstance(obj, (int, float, complex, str, bytes)) or obj is None:
+        yield repr(obj)
+    else:
+        raise TypeError('not handled: ', type(obj))
+
+
+class CachingWrapper:
+    """This object wraps a callable and caches the results in a dbm database on the file system (by default in the temp
+    folder). The key is generated via to_key which means that large arguments need a large time to process."""
+
+    def __init__(self, func, storage_path=None):
+        self._func = func
+
+        if storage_path is None:
+            storage_path = os.path.join(tempfile.gettempdir(), inspect.getmodule(func).__name__ + func.__name__)
+
+        self.storage_path = storage_path
+
+    def __call__(self, *args, **kwargs):
+        key = ''.join(to_key((args, kwargs)))
+
+        with shelve.open(self.storage_path) as db:
+            if key in db:
+                result = db[key]
+            else:
+                result = self._func(*args, **kwargs)
+                db[key] = result
+        return result
+
+    def clear(self):
+        with shelve.open(self.storage_path) as db:
+            db.clear()
+
+
+def file_cache(func: Callable) -> Callable:
+    return CachingWrapper(func)
+
+
+lru_cache = functools.lru_cache