mirror of
https://github.com/wassname/cachier.git
synced 2026-09-09 11:19:08 +08:00
clearer seperation of functionalities into sub-modules
This commit is contained in:
Shay Palachy
@@ -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 <shaypal5@gmail.com>
|
||||
|
||||
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."""
|
||||
+5
-347
@@ -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):
|
||||
|
||||
@@ -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 <shaypal5@gmail.com>
|
||||
|
||||
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}
|
||||
}
|
||||
)
|
||||
@@ -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 <shaypal5@gmail.com>
|
||||
|
||||
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)
|
||||
Reference in New Issue
Block a user