其他
只用两个函数实现事务的设计模式!
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 restore
Transaction对象
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 wrapper
class 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 transaction
class 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 如何落地?揭秘阿里核心业务大规模落地实现