python - 使用 contextmanager 捕获指令以供以后执行

Question

我想使用上下文管理器实现类似伪数据库的事务。

举个例子:

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. 所有排队的调用必须是基于您的“ session ”实例的方法。其他任何事情都会立即执行。
2. 我不知道您想如何处理不可调用的 session 属性，所以现在我假设它只会检索值。

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

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)
# ]

由于您希望它的结构方式，此解决方案有点做作，但如果您需要上下文管理器并且需要 session 看起来像一个直接的函数调用，只使用 partial 是微不足道的:

my_queue = []

# some session
my_queue.append(partial(f, a))
my_queue.append(partial(f, b))
for func in my_queue:
    func()