diff --git a/cachier/base_core.py b/cachier/base_core.py new file mode 100644 index 0000000..1bc73a3 --- /dev/null +++ b/cachier/base_core.py @@ -0,0 +1,57 @@ +"""Defines the interface of a cachier caching core.""" +# This file is part of Cachier. +# https://github.com/shaypal5/cachier + +# Licensed under the MIT license: +# http://www.opensource.org/licenses/MIT-license +# Copyright (c) 2016, Shay Palachy + +import abc # for the _BaseCore abstract base class + +class _BaseCore(object): + __metaclass__ = abc.ABCMeta + + def __init__(self, stale_after, next_time): + self.stale_after = stale_after + self.next_time = next_time + self.func = None + + def set_func(self, func): + """Sets the function this core will use. This has to be set before + any method is called""" + self.func = func + + @abc.abstractmethod + def get_entry_by_key(self, key): + """Returns the result mapped to the given key in this core's cache, + if such a mapping exists.""" + + @abc.abstractmethod + def get_entry(self, args, kwds): + """Returns the result mapped to the given arguments in this core's + cache, if such a mapping exists.""" + + @abc.abstractmethod + def set_entry(self, key, func_res): + """Maps the given result to the given key in this core's cache.""" + + @abc.abstractmethod + def mark_entry_being_calculated(self, key): + """Marks the entry mapped by the given key as being calculated.""" + + @abc.abstractmethod + def mark_entry_not_calculated(self, key): + """Marks the entry mapped by the given key as not being calculated.""" + + @abc.abstractmethod + def wait_on_entry_calc(self, key): + """Waits on the entry mapped by key being calculated and returns the + result.""" + + @abc.abstractmethod + def clear_cache(self): + """Clears the cache of this core.""" + + @abc.abstractmethod + def clear_being_calculated(self): + """Marks all entries in this cache as not being calculated.""" diff --git a/cachier/core.py b/cachier/core.py index d78b78f..6f0915c 100644 --- a/cachier/core.py +++ b/cachier/core.py @@ -9,9 +9,8 @@ import os from functools import wraps -import pickle # for local caching + import datetime -import abc # for the _BaseCore abstract base class try: # for asynchronous file uploads from concurrent.futures import ThreadPoolExecutor except ImportError: # we're in python 2.x @@ -24,354 +23,14 @@ except ImportError: # we're in python 2.x if 'futures' not in PACKAGES: pip.main(['install', 'futures']) from concurrent.futures import ThreadPoolExecutor -import time # to sleep when waiting on Mongo cache -import fcntl # to lock on pickle cache IO -import pymongo -from bson.binary import Binary # to save binary data to mongodb -from watchdog.observers import Observer -from watchdog.events import PatternMatchingEventHandler +from .pickle_core import _PickleCore +from .mongo_core import _MongoCore + -CACHIER_DIR = '~/.cachier/' -EXPANDED_CACHIER_DIR = os.path.expanduser(CACHIER_DIR) DEFAULT_MAX_WORKERS = 8 -MONGO_SLEEP_DURATION_IN_SEC = 6 -# === Cores definitions === - -class _BaseCore(object): - __metaclass__ = abc.ABCMeta - - def __init__(self, stale_after, next_time): - self.stale_after = stale_after - self.next_time = next_time - self.func = None - - def set_func(self, func): - """Sets the function this core will use. This has to be set before - any method is called""" - self.func = func - - @abc.abstractmethod - def get_entry_by_key(self, key): - """Returns the result mapped to the given key in this core's cache, - if such a mapping exists.""" - - @abc.abstractmethod - def get_entry(self, args, kwds): - """Returns the result mapped to the given arguments in this core's - cache, if such a mapping exists.""" - - @abc.abstractmethod - def set_entry(self, key, func_res): - """Maps the given result to the given key in this core's cache.""" - - @abc.abstractmethod - def mark_entry_being_calculated(self, key): - """Marks the entry mapped by the given key as being calculated.""" - - @abc.abstractmethod - def mark_entry_not_calculated(self, key): - """Marks the entry mapped by the given key as not being calculated.""" - - @abc.abstractmethod - def wait_on_entry_calc(self, key): - """Waits on the entry mapped by key being calculated and returns the - result.""" - - @abc.abstractmethod - def clear_cache(self): - """Clears the cache of this core.""" - - @abc.abstractmethod - def clear_being_calculated(self): - """Marks all entries in this cache as not being calculated.""" - - -class _MongoCore(_BaseCore): - - def __init__(self, mongetter, stale_after, next_time): - _BaseCore.__init__(self, stale_after, next_time) - self.mongetter = mongetter - self.mongo_collection = None - - @staticmethod - def _get_func_str(func): - return '.{}.{}'.format(func.__module__, func.__name__) - - def _get_mongo_collection(self): - if not self.mongo_collection: - self.mongo_collection = self.mongetter() - return self.mongo_collection - - def get_entry_by_key(self, key): - res = self._get_mongo_collection().find_one({ - 'func': _MongoCore._get_func_str(self.func), - 'key': key - }) - if res: - try: - entry = { - 'value': pickle.loads(res['value']), - 'time': res.get('time', None), - 'stale': res.get('stale', False), - 'being_calculated': res.get('being_calculated', False) - } - except KeyError: - entry = { - 'value': None, - 'time': res.get('time', None), - 'stale': res.get('stale', False), - 'being_calculated': res.get('being_calculated', False) - } - return key, entry - return key, None - - def get_entry(self, args, kwds): - key = pickle.dumps(args + tuple(sorted(kwds.items()))) - # print('key type={}, key={}'.format( - # type(key), key)) - return self.get_entry_by_key(key) - - def set_entry(self, key, func_res): - thebytes = pickle.dumps(func_res) - self._get_mongo_collection().update_one( - { - 'func': _MongoCore._get_func_str(self.func), - 'key': key - }, - { - '$set': { - 'func': _MongoCore._get_func_str(self.func), - 'key': key, - 'value': Binary(thebytes), - 'time': datetime.datetime.now(), - 'stale': False, - 'being_calculated': False - } - }, - upsert=True - ) - - def mark_entry_being_calculated(self, key): - self._get_mongo_collection().update_one( - { - 'func': _MongoCore._get_func_str(self.func), - 'key': key - }, - { - '$set': {'being_calculated': True} - }, - upsert=True - ) - - def mark_entry_not_calculated(self, key): - try: - self._get_mongo_collection().update_one( - { - 'func': _MongoCore._get_func_str(self.func), - 'key': key - }, - { - '$set': {'being_calculated': False} - }, - upsert=False # should not insert in this case - ) - except pymongo.errors.OperationFailure: - pass # don't care in this case - - def wait_on_entry_calc(self, key): - while True: - time.sleep(MONGO_SLEEP_DURATION_IN_SEC) - key, entry = self.get_entry_by_key(key) - if entry is not None and not entry['being_calculated']: - return entry['value'] - # key, entry = self.get_entry_by_key(key) - # if entry is not None: - # return entry['value'] - # return None - - def clear_cache(self): - self._get_mongo_collection().delete_many( - {'func': _MongoCore._get_func_str(self.func)}) - - def clear_being_calculated(self): - self._get_mongo_collection().update_many( - { - 'func': _MongoCore._get_func_str(self.func), - 'being_calculated': True - }, - { - '$set': {'being_calculated': False} - } - ) - -class _PickleCore(_BaseCore): - - class CacheChangeHandler(PatternMatchingEventHandler): - """Handles cache-file modification events.""" - - def __init__(self, filename, core, key): - PatternMatchingEventHandler.__init__( - self, - patterns=["*" + filename], - ignore_patterns=None, - ignore_directories=True, - case_sensitive=False - ) - self.core = core - self.key = key - self.observer = None - self.value = None - - def inject_observer(self, observer): - """Inject the observer running this handler.""" - self.observer = observer - - def _check_calculation(self): - # print('checking calc') - entry = self.core.get_entry_by_key(self.key, True)[1] - # print(self.key) - # print(entry) - if not entry['being_calculated']: - # print('stoping observer!') - self.value = entry['value'] - self.observer.stop() - # print('NOT stoping observer... :(') - - def on_created(self, event): - self._check_calculation() - - def on_modified(self, event): - self._check_calculation() - - def __init__(self, stale_after, next_time, reload): - _BaseCore.__init__(self, stale_after, next_time) - self.cache = None - self.reload = reload - - def _get_cache_file_name(self): - return '.{}.{}'.format( - self.func.__module__, self.func.__name__) # pylint: disable=W0212 - - def _get_cache_path(self): - # print(EXPANDED_CACHIER_DIR) - if not os.path.exists(EXPANDED_CACHIER_DIR): - os.makedirs(EXPANDED_CACHIER_DIR) - fpath = os.path.abspath(os.path.join( - os.path.realpath(EXPANDED_CACHIER_DIR), - self._get_cache_file_name() - )) - # print(fpath) - return fpath - - def _reload_cache(self): - fpath = self._get_cache_path() - try: - with open(fpath, 'rb') as cache_file: - fcntl.flock(cache_file, fcntl.LOCK_SH) - try: - self.cache = pickle.load(cache_file) - except EOFError: - self.cache = {} - fcntl.flock(cache_file, fcntl.LOCK_UN) - except FileNotFoundError: - self.cache = {} - - def _get_cache(self): - if not self.cache: - self._reload_cache() - return self.cache - - def _save_cache(self, cache): - self.cache = cache - fpath = self._get_cache_path() - with open(fpath, 'wb') as cache_file: - fcntl.flock(cache_file, fcntl.LOCK_EX) - pickle.dump(cache, cache_file) - fcntl.flock(cache_file, fcntl.LOCK_UN) - self._reload_cache() - - def get_entry_by_key(self, key, reload=False): # pylint: disable=W0221 - # print('{}, {}'.format(self.reload, reload)) - if self.reload or reload: - self._reload_cache() - return key, self._get_cache().get(key, None) - - def get_entry(self, args, kwds): - key = args + tuple(sorted(kwds.items())) - # print('key type={}, key={}'.format(type(key), key)) - return self.get_entry_by_key(key) - - def set_entry(self, key, func_res): - cache = self._get_cache() - cache[key] = { - 'value': func_res, - 'time': datetime.datetime.now(), - 'stale': False, - 'being_calculated': False - } - self._save_cache(cache) - - def mark_entry_being_calculated(self, key): - cache = self._get_cache() - try: - cache[key]['being_calculated'] = True - except KeyError: - cache[key] = { - 'value': None, - 'time': datetime.datetime.now(), - 'stale': False, - 'being_calculated': True - } - self._save_cache(cache) - - def mark_entry_not_calculated(self, key): - cache = self._get_cache() - try: - cache[key]['being_calculated'] = False - self._save_cache(cache) - except KeyError: - pass # that's ok, we don't need an entry in that case - - - def wait_on_entry_calc(self, key): - entry = self._get_cache()[key] - if not entry['being_calculated']: - return entry['value'] - event_handler = _PickleCore.CacheChangeHandler( - filename=self._get_cache_file_name(), - core=self, - key=key - ) - observer = Observer() - event_handler.inject_observer(observer) - observer.schedule( - event_handler, - path=EXPANDED_CACHIER_DIR, - recursive=True - ) - observer.start() - observer.join(timeout=2.0) - if observer.isAlive(): - # print('Timedout waiting. Starting again...') - return self.wait_on_entry_calc(key) - # print("Returned value: {}".format(event_handler.value)) - return event_handler.value - - def clear_cache(self): - self._save_cache({}) - - def clear_being_calculated(self): - cache = self._get_cache() - for key in cache: - cache[key]['being_calculated'] = False - self._save_cache(cache) - - -# === Main functionality === - def _max_workers(): try: return int(os.environ['CACHIER_MAX_WORKERS']) @@ -402,8 +61,7 @@ def _function_thread(core, key, func, args, kwds): except BaseException as exc: # pylint: disable=W0703 print( 'Function call failed with the following exception:\n{}'.format( - exc) - ) + exc)) def _calc_entry(core, key, func, args, kwds): diff --git a/cachier/mongo_core.py b/cachier/mongo_core.py new file mode 100644 index 0000000..da8d1e0 --- /dev/null +++ b/cachier/mongo_core.py @@ -0,0 +1,148 @@ +"""A MongoDB-based caching core for cachier.""" + +# This file is part of Cachier. +# https://github.com/shaypal5/cachier + +# Licensed under the MIT license: +# http://www.opensource.org/licenses/MIT-license +# Copyright (c) 2016, Shay Palachy + +import pickle # for serialization of python objects +from datetime import datetime +import time # to sleep when waiting on Mongo cache + +from pymongo import ( + IndexModel, + ASCENDING +) +from pymongo.errors import OperationFailure +from bson.binary import Binary # to save binary data to mongodb + +from .base_core import _BaseCore + + +MONGO_SLEEP_DURATION_IN_SEC = 6 + + +class _MongoCore(_BaseCore): + + def __init__(self, mongetter, stale_after, next_time): + _BaseCore.__init__(self, stale_after, next_time) + self.mongetter = mongetter + self.mongo_collection = self.mongetter() + if '_func_1_key_1' not in self.mongo_collection.index_information(): + func1key1 = IndexModel( + [('func', ASCENDING), ('key', ASCENDING)], + name='_func_1_key_1') + self.mongo_collection.create_indexes([func1key1]) + + @staticmethod + def _get_func_str(func): + return '.{}.{}'.format(func.__module__, func.__name__) + + def _get_mongo_collection(self): + if not self.mongo_collection: + self.mongo_collection = self.mongetter() + return self.mongo_collection + + def get_entry_by_key(self, key): + res = self._get_mongo_collection().find_one({ + 'func': _MongoCore._get_func_str(self.func), + 'key': key + }) + if res: + try: + entry = { + 'value': pickle.loads(res['value']), + 'time': res.get('time', None), + 'stale': res.get('stale', False), + 'being_calculated': res.get('being_calculated', False) + } + except KeyError: + entry = { + 'value': None, + 'time': res.get('time', None), + 'stale': res.get('stale', False), + 'being_calculated': res.get('being_calculated', False) + } + return key, entry + return key, None + + def get_entry(self, args, kwds): + key = pickle.dumps(args + tuple(sorted(kwds.items()))) + # print('key type={}, key={}'.format( + # type(key), key)) + return self.get_entry_by_key(key) + + def set_entry(self, key, func_res): + thebytes = pickle.dumps(func_res) + self._get_mongo_collection().update_one( + { + 'func': _MongoCore._get_func_str(self.func), + 'key': key + }, + { + '$set': { + 'func': _MongoCore._get_func_str(self.func), + 'key': key, + 'value': Binary(thebytes), + 'time': datetime.now(), + 'stale': False, + 'being_calculated': False + } + }, + upsert=True + ) + + def mark_entry_being_calculated(self, key): + self._get_mongo_collection().update_one( + { + 'func': _MongoCore._get_func_str(self.func), + 'key': key + }, + { + '$set': {'being_calculated': True} + }, + upsert=True + ) + + def mark_entry_not_calculated(self, key): + try: + self._get_mongo_collection().update_one( + { + 'func': _MongoCore._get_func_str(self.func), + 'key': key + }, + { + '$set': {'being_calculated': False} + }, + upsert=False # should not insert in this case + ) + except OperationFailure: + pass # don't care in this case + + def wait_on_entry_calc(self, key): + while True: + time.sleep(MONGO_SLEEP_DURATION_IN_SEC) + key, entry = self.get_entry_by_key(key) + if entry is not None and not entry['being_calculated']: + return entry['value'] + # key, entry = self.get_entry_by_key(key) + # if entry is not None: + # return entry['value'] + # return None + + def clear_cache(self): + self._get_mongo_collection().delete_many( + {'func': _MongoCore._get_func_str(self.func)}) + + def clear_being_calculated(self): + self._get_mongo_collection().update_many( + { + 'func': _MongoCore._get_func_str(self.func), + 'being_calculated': True + }, + { + '$set': {'being_calculated': False} + } + ) diff --git a/cachier/pickle_core.py b/cachier/pickle_core.py new file mode 100644 index 0000000..e299369 --- /dev/null +++ b/cachier/pickle_core.py @@ -0,0 +1,184 @@ +"""A pickle-based caching core for cachier.""" + +# This file is part of Cachier. +# https://github.com/shaypal5/cachier + +# Licensed under the MIT license: +# http://www.opensource.org/licenses/MIT-license +# Copyright (c) 2016, Shay Palachy + +import os +import pickle # for local caching +import fcntl # to lock on pickle cache IO +from datetime import datetime + +from watchdog.observers import Observer +from watchdog.events import PatternMatchingEventHandler + +from .base_core import _BaseCore + + +CACHIER_DIR = '~/.cachier/' +EXPANDED_CACHIER_DIR = os.path.expanduser(CACHIER_DIR) + + +class _PickleCore(_BaseCore): + + class CacheChangeHandler(PatternMatchingEventHandler): + """Handles cache-file modification events.""" + + def __init__(self, filename, core, key): + PatternMatchingEventHandler.__init__( + self, + patterns=["*" + filename], + ignore_patterns=None, + ignore_directories=True, + case_sensitive=False + ) + self.core = core + self.key = key + self.observer = None + self.value = None + + def inject_observer(self, observer): + """Inject the observer running this handler.""" + self.observer = observer + + def _check_calculation(self): + # print('checking calc') + entry = self.core.get_entry_by_key(self.key, True)[1] + # print(self.key) + # print(entry) + if not entry['being_calculated']: + # print('stoping observer!') + self.value = entry['value'] + self.observer.stop() + # print('NOT stoping observer... :(') + + def on_created(self, event): + self._check_calculation() + + def on_modified(self, event): + self._check_calculation() + + def __init__(self, stale_after, next_time, reload): + _BaseCore.__init__(self, stale_after, next_time) + self.cache = None + self.reload = reload + + def _get_cache_file_name(self): + return '.{}.{}'.format( + self.func.__module__, self.func.__name__) # pylint: disable=W0212 + + def _get_cache_path(self): + # print(EXPANDED_CACHIER_DIR) + if not os.path.exists(EXPANDED_CACHIER_DIR): + os.makedirs(EXPANDED_CACHIER_DIR) + fpath = os.path.abspath(os.path.join( + os.path.realpath(EXPANDED_CACHIER_DIR), + self._get_cache_file_name() + )) + # print(fpath) + return fpath + + def _reload_cache(self): + fpath = self._get_cache_path() + try: + with open(fpath, 'rb') as cache_file: + fcntl.flock(cache_file, fcntl.LOCK_SH) + try: + self.cache = pickle.load(cache_file) + except EOFError: + self.cache = {} + fcntl.flock(cache_file, fcntl.LOCK_UN) + except FileNotFoundError: + self.cache = {} + + def _get_cache(self): + if not self.cache: + self._reload_cache() + return self.cache + + def _save_cache(self, cache): + self.cache = cache + fpath = self._get_cache_path() + with open(fpath, 'wb') as cache_file: + fcntl.flock(cache_file, fcntl.LOCK_EX) + pickle.dump(cache, cache_file) + fcntl.flock(cache_file, fcntl.LOCK_UN) + self._reload_cache() + + def get_entry_by_key(self, key, reload=False): # pylint: disable=W0221 + # print('{}, {}'.format(self.reload, reload)) + if self.reload or reload: + self._reload_cache() + return key, self._get_cache().get(key, None) + + def get_entry(self, args, kwds): + key = args + tuple(sorted(kwds.items())) + # print('key type={}, key={}'.format(type(key), key)) + return self.get_entry_by_key(key) + + def set_entry(self, key, func_res): + cache = self._get_cache() + cache[key] = { + 'value': func_res, + 'time': datetime.now(), + 'stale': False, + 'being_calculated': False + } + self._save_cache(cache) + + def mark_entry_being_calculated(self, key): + cache = self._get_cache() + try: + cache[key]['being_calculated'] = True + except KeyError: + cache[key] = { + 'value': None, + 'time': datetime.now(), + 'stale': False, + 'being_calculated': True + } + self._save_cache(cache) + + def mark_entry_not_calculated(self, key): + cache = self._get_cache() + try: + cache[key]['being_calculated'] = False + self._save_cache(cache) + except KeyError: + pass # that's ok, we don't need an entry in that case + + def wait_on_entry_calc(self, key): + entry = self._get_cache()[key] + if not entry['being_calculated']: + return entry['value'] + event_handler = _PickleCore.CacheChangeHandler( + filename=self._get_cache_file_name(), + core=self, + key=key + ) + observer = Observer() + event_handler.inject_observer(observer) + observer.schedule( + event_handler, + path=EXPANDED_CACHIER_DIR, + recursive=True + ) + observer.start() + observer.join(timeout=2.0) + if observer.isAlive(): + # print('Timedout waiting. Starting again...') + return self.wait_on_entry_calc(key) + # print("Returned value: {}".format(event_handler.value)) + return event_handler.value + + def clear_cache(self): + self._save_cache({}) + + def clear_being_calculated(self): + cache = self._get_cache() + for key in cache: + cache[key]['being_calculated'] = False + self._save_cache(cache)