使用contextmanager捕获指令以便以后执行

Question

我想使用上下文管理器实现类似db的事务。

举个例子：

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，我想立即执行指令。

Answer 1

它并不漂亮，但要遵循您正在寻找的结构，您必须构建一个临时事务类，它保存您的函数队列并在上下文管理器退出后执行它。你需要使用functools.partial，但是有一些限制：

1. 所有排队的调用必须是基于“会话”实例的方法。其他任何事情都会马上执行。
2. 我不知道你想如何处理不可调用的会话属性，所以现在我假设它只是检索值。

话虽如此，这是我的看法：

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()