-
Notifications
You must be signed in to change notification settings - Fork 1
/
model_cache.py
47 lines (39 loc) · 1.84 KB
/
model_cache.py
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
# -*- coding: utf-8 -*-
from .model_cache_class import *
valid_storage_types = ("memory", "sqlite", "redis")
class ModelCache():
@classmethod
def connect(cls, original_model, **kwargs):
# assert original_model's behavior
process_notifier(original_model)
# setup args
default_kwargs = {
'cache_dir' : None,
'storage_type' : 'sqlite',
'percentage' : 0.9999,
'filter_lambda' : lambda item1: False,
'read_id_lambda' : lambda item1: str(item1['_id']),
'included_class' : object,
}
for k1, v1 in kwargs.iteritems():
if k1 in default_kwargs:
default_kwargs[k1] = v1
# validate storage
assert default_kwargs['storage_type'] in valid_storage_types
if (default_kwargs['cache_dir'] is None) and (default_kwargs['storage_type'] != "memory"):
raise Exception(u"`cache_dir` should not be None when storage_type is not memory.")
# decorate class
def _model_cache_decorator(decorated_class):
# ensure decorated_class's methods will overwrite ModelCache's.
class _model_cache(decorated_class, default_kwargs['included_class'], ModelCacheClass):
class OriginalClass(): pass # so we can setattr here.
original = OriginalClass()
for k1, v1 in default_kwargs.iteritems():
setattr(original, k1, v1)
del k1; del v1
original.model = original_model
_model_cache.__name__ = decorated_class.__name__
_model_cache.__module__ = decorated_class.__module__ # so can pickle :)
_model_cache.connect()
return _model_cache
return _model_cache_decorator