Files
@ 05c5cfcb4e8e
Branch filter:
Location: symposion_app/registrasion/controllers/batch.py
05c5cfcb4e8e
3.4 KiB
text/x-python
Adds first tests for automatic credit note application
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 | import contextlib
import functools
from django.contrib.auth.models import User
class BatchController(object):
''' Batches are sets of operations where certain queries for users may be
repeated, but are also unlikely change within the boundaries of the batch.
Batches are keyed per-user. You can mark the edge of the batch with the
``batch`` context manager. If you nest calls to ``batch``, only the
outermost call will have the effect of ending the batch.
Batches store results for functions wrapped with ``memoise``. These results
for the user are flushed at the end of the batch.
If a return for a memoised function has a callable attribute called
``end_batch``, that attribute will be called at the end of the batch.
'''
_user_caches = {}
_NESTING_KEY = "nesting_count"
@classmethod
@contextlib.contextmanager
def batch(cls, user):
''' Marks the entry point for a batch for the given user. '''
cls._enter_batch_context(user)
try:
yield
finally:
# Make sure we clean up in case of errors.
cls._exit_batch_context(user)
@classmethod
def _enter_batch_context(cls, user):
if user not in cls._user_caches:
cls._user_caches[user] = cls._new_cache()
cache = cls._user_caches[user]
cache[cls._NESTING_KEY] += 1
@classmethod
def _exit_batch_context(cls, user):
cache = cls._user_caches[user]
cache[cls._NESTING_KEY] -= 1
if cache[cls._NESTING_KEY] == 0:
cls._call_end_batch_methods(user)
del cls._user_caches[user]
@classmethod
def _call_end_batch_methods(cls, user):
cache = cls._user_caches[user]
ended = set()
while True:
keys = set(cache.keys())
if ended == keys:
break
keys_to_end = keys - ended
for key in keys_to_end:
item = cache[key]
if hasattr(item, 'end_batch') and callable(item.end_batch):
item.end_batch()
ended = ended | keys_to_end
@classmethod
def memoise(cls, func):
''' Decorator that stores the result of the stored function in the
user's results cache until the batch completes. Keyword arguments are
not yet supported.
Arguments:
func (callable(*a)): The function whose results we want
to store. The positional arguments, ``a``, are used as cache
keys.
Returns:
callable(*a): The memosing version of ``func``.
'''
@functools.wraps(func)
def f(*a):
for arg in a:
if isinstance(arg, User):
user = arg
break
else:
raise ValueError("One position argument must be a User")
func_key = (func, tuple(a))
cache = cls.get_cache(user)
if func_key not in cache:
cache[func_key] = func(*a)
return cache[func_key]
return f
@classmethod
def get_cache(cls, user):
if user not in cls._user_caches:
# Return blank cache here, we'll just discard :)
return cls._new_cache()
return cls._user_caches[user]
@classmethod
def _new_cache(cls):
''' Returns a new cache dictionary. '''
cache = {}
cache[cls._NESTING_KEY] = 0
return cache
|