-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathfuncutils.py
79 lines (67 loc) · 1.92 KB
/
funcutils.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
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
from functools import wraps
def caching(f):
@wraps(f)
def decorated(self, *args, **kws):
key = '_cached_' + f.__name__
result = getattr(self, key, None)
if result is None:
result = f(self, *args, **kws)
setattr(self, key, result)
return result
return decorated
def getattrs(obj, attrnames, default=None):
attrs = attrnames.split(".") if isinstance(attrnames, basestring) else attrnames
return _getattrs(obj, attrs, default)
def _getattrs(obj, attrs, default):
if not attrs:
return obj
attr = attrs[0]
if attr.startswith("*"):
attr = attr[1:]
return _getattrs([
getattr(x, attr) for x in obj if hasattr(x,attr)], attrs[1:], default)
return [y for x in obj
for y in getattr(x, attr, ())]
if hasattr(obj, attr):
return _getattrs(getattr(obj, attr), attrs[1:], default)
it = getattr(obj, '__iter__', None)
if it:
return _getattrs([
getattr(x, attr) for x in it() if hasattr(x,attr)], attrs[1:], default)
return default
def setattrs(obj, d):
for k,v in d.items():
setattr(obj,k,v)
def keyword_only(f):
@wraps(f)
def decorated(self, **kws):
try:
return f(self,**kws)
except:
raise
decorated = getattr(f,'original', f)
return decorated
from contextlib import contextmanager
@contextmanager
def transaction(session):
try:
#session.begin()
session.extension.after_begin(session, None,None)
yield session
session.flush()
except:
session.rollback()
raise
else:
session.commit()
finally:
session.remove()
def run(f):
return f()
def run_with(context):
def deco(f):
with context as x:
return f(x)
return deco
def run_with_transaction(session):
return run_with(transaction(session))