python - 使用 contextmanager 捕获指令以供以后执行
问题描述
我想使用上下文管理器实现类似伪数据库的事务。
举个例子:
class Transactor:
def a(): pass
def b(d, b): pass
def c(i): pass
@contextmanager
def get_session(self):
txs = []
yield self # accumulate method calls
for tx in tx:
tx() # somehow pass the arguments
def main():
t = Transactor()
with t.get_session() as session:
session.a() # inserts `a` into `txs`
... more code ...
session.c(value) # inserts `c` and `(value)` into `txs`
session.b(value1, value2) # inserts `b` and `(value1, value2)` into `txs`
... more code ...
# non-transator related code
f = open('file.txt') # If this throws an exception,
# break out of the context manager,
# and discard previous transactor calls.
... more code ...
session.a() # inserts `a` into `txs`
session.b(x, y) # inserts `b` and `(x, y)` into `txs`
# Now is outside of context manager.
# The following calls should execute immediately
t.a()
t.b(x, y)
t.c(k)
如果出现异常等问题,则丢弃txs
(回滚)。如果它到达上下文的末尾,则按插入顺序执行每条指令并传入适当的参数。
如何捕获方法调用以供以后执行?
还有一个额外的警告:如果get_session
没有调用,我想立即执行指令。
解决方案
这并不漂亮,但要遵循您正在寻找的结构,您必须构建一个临时事务类来保存您的函数队列并在上下文管理器退出后执行它。您需要使用functools.partial
,但有一些限制:
- 所有排队的调用必须是基于您的“会话”实例的方法。其他任何事情都会立即执行。
- 我不知道你想如何处理不可调用的会话属性,所以现在我假设它只会检索值。
话虽如此,这是我的看法:
from functools import partial
class TempTrans:
# pass in the object instance to mimic
def __init__(self, obj):
self._queue = []
# iterate through the attributes and methods within the object and its class
for attr, val in type(obj).__dict__.items() ^ obj.__dict__.items():
if not attr.startswith('_'):
if callable(val):
setattr(self, attr, partial(self._add, getattr(obj, attr)))
else:
# placeholder to handle non-callable attributes
setattr(self, attr, val)
# function to add to queue
def _add(self, func, *args, **kwargs):
self._queue.append(partial(func, *args, **kwargs))
# function to execute the queue
def _execute(self):
_remove = []
# iterate through the queue to call the functions.
# I suggest catching errors here in case your functions falls through
for func in self._queue:
try:
func()
_remove.append(func)
except Exception as e:
print('some error occured')
break
# remove the functions that were successfully ran
for func in _remove:
self._queue.remove(func)
现在进入上下文管理器(它将在您的类之外,如果您愿意,可以将其作为类方法放入):
@contextmanager
def temp_session(obj):
t = TempTrans(obj)
try:
yield t
t._execute()
print('Transactions successfully ran')
except:
print('Encountered errors, queue was not executed')
finally:
print(t._queue) # debug to see what's left of the queue
用法:
f = Foo()
with temp_session(f) as session:
session.a('hello')
session.b(1, 2, 3)
# a hello
# b 1 2 3
# Transactions successfully ran
# []
with temp_session(f) as session:
session.a('hello')
session.b(1, 2, 3)
session.attrdoesnotexist # expect an error
# Encountered errors, queue was not executed
# [
# functools.partial(<bound method Foo.a of <__main__.Foo object at 0x0417D3B0>>, 'hello'),
# functools.partial(<bound method Foo.b of <__main__.Foo object at 0x0417D3B0>>, 1, 2, 3)
# ]
由于您希望它的结构方式,这个解决方案有点做作,但是如果您不需要上下文管理器并且不需要会话看起来像直接函数调用,那么使用它是微不足道的partial
:
my_queue = []
# some session
my_queue.append(partial(f, a))
my_queue.append(partial(f, b))
for func in my_queue:
func()
推荐阅读
- c# - 我的客户端的“如果”语句被我的函数忽略,即使它成立
- java - 无论Java中的内容类型如何,如何将字段作为字符串获取?
- excel - VBA CopyFolder复制错误的文件夹
- guidewire - 在 PCF 的下拉列表中过滤掉用户
- php - 具有继承和类型提示的静态工厂模式
- android - 当我尝试构建我的 React Native App 的签名 APK 时出现 2 个错误
- c++ - 简单快速的读取过程
- nestjs - 覆盖 GlobalPipes
- tensorflow - data.make_initializable_iterator() 抛出错误:TypeFetch 参数必须是字符串或张量
- typescript - 我应该为 Vue 中的类组件使用哪些访问修饰符?