其他
只用两个函数实现事务的设计模式！
The following article is from TechFlow Author 梁唐
深度拷贝
memento
from copy import copy, deepcopy
def memento(obj, deep=False): state = deepcopy(obj.__dict__) if deep else copy(obj.__dict__)
def restore(): obj.__dict__.clear() obj.__dict__.update(state) return restoreTransaction对象
class Transaction:
deep = False states = []
def __init__(self, deep, *targets): self.deep = deep self.targets = targets self.commit()
def commit(self): self.states = [memento(target, self.deep) for target in self.targets]
def rollback(self): for a_state in self.states: a_state()Transaction装饰器
from functools import wraps
def transactional(func): @wraps(func) def wrapper(*args, **kwargs): # args[0] is obj state = memento(args[0]) try: func(*args, **kwargs) except Exception as e: state() raise e return wrapperclass Transactional:
def __init__(self, method): self.method = method
def __get__(self, obj, cls): def transaction(*args, **kwargs): state = memento(obj) try: return self.method(*args, **kwargs) except Exception as e: state() raise e return transactionclass Wrapper: def __init__(self, func): wraps(func)(self)
def __call__(self, *args, **kwargs): return self.__wrapped__(*args, **kwargs)
def __get__(self, instance, cls): if instance is None: return self else: return types.MethodType(self, instance)实战
class NumObj: def __init__(self, value): self.value = value
def __repr__(self): return '<%s, %r>' % (self.__class__.__name__, self.value)
def increment(self): self.value += 1
@transactional def do_stuff(self): self.value += '111' self.increment() if __name__ == '__main__': num_obj = NumObj(-1)
a_transaction = Transaction(True, num_obj) # 使用Transaction try: for i in range(3): num_obj.increment() print(num_obj)
a_transaction.commit() print('----committed') for i in range(3): num_obj.increment() print(num_obj) num_obj.value += 'x' print(num_obj) except Exception: a_transaction.rollback() print('----rollback')
print(num_obj) # 使用Transactional print('-- now doing stuff') num_obj.increment()
try: num_obj.do_stuff() except Exception: print('-> doing stuff failed') import sys import traceback traceback.print_exc(file=sys.stdout)
print(num_obj)更多精彩推荐
☞挑战 Linux 之父认为的“不可能”：向 M1 Mac 移植 Linux
☞酷派奖励程序员10 万股期权！因代码贡献受 Linux 之父亲自点名赞赏
☞在英雄联盟地图中寻找“数据结构的大门”
☞Serverless 如何落地？揭秘阿里核心业务大规模落地实现