python3/contextlib: 将@contextlib.contextmanager 转换为 acquire/release？

Question

我正在使用一些第 3 方代码，它可以通过装饰有 @contextlib.contextmanager 的例程启用锁定。我还使用大型 python3 代码库，我们可以在其中插入不同的锁定软件，只要我能够实现 acquire 和 release 方法。

我正在尝试在此软件结构中使用第 3 方代码（不知道它是如何编写的）。

为了阐明我在寻找什么，假设第 3 方锁定例程之一被编写为标准 @contextlib.contextmanager 生成器，如下所示：

@contextlib.contextmanager
def lock(arg0, arg1):
    try:
        # This section of code corresponds to `acquire`.
        # Acquire a lock called 'lock', and then ...
        yield lock
    finally:
        # This section of code coresponds to `release`.
        # Do cleanup.

通常会这样使用...

with third.party.lock(arg0, arg1):
    # Do stuff in this critical section

但是正如我上面提到的，我想写一个 class，它有一个 acquire 方法和一个 release 方法，它使用 third.party.lock，我我想通过现有的 third.party.lock 模块来完成它，而不是重写它。

换句话说，我想写一个class，看起来像这样...

class LockWrapper(object):

    def __init__(self):
        # initialization

    def acquire(self):
        # Use third.party.lock to obtain a lock.
        # ??? ??? ???

    def release(self):
        # I don't know what to do here. There is no yield
        # in the `finally` section of a normal
        # @contextlib.contextmanager decorated method/
        # ??? ??? ???

正如我在示例代码的注释中所述，我看不出如何让 acquire 和 release 做任何有意义的事情。

看起来我必须从原始 third.party.lock 模块中窃取代码才能完成此操作，但我希望我忽略了一种无需了解任何相关内容即可执行此操作的方法这个第 3 方代码。

我运气不好吗？

非常感谢。

Answer 1

好的，我明白了。我的回答基于这里的一些代码：https://gist.github.com/icio/c0d3f7efd415071f725b

关键是这个enter_context辅助函数，它深入到上下文管理器的结构中。它类似于该站点上同名的函数...

def enter_context(func, *args, **kwargs):
    def _acquire():
        with func(*args, **kwargs) as f:
            yield f
    acquire_gen = _acquire()
    def release_func():
        try:
            next(acquire_gen)
        except StopIteration:
            pass
    return acquire_gen, release_func

以下 class 为传递给构造函数的任何上下文管理器函数实现 acquire 和 release。我将 enter_context 封装在这个 class ...

中

class ContextWrapper(object):
    @staticmethod
    def enter_context(func, *args, **kwargs):
        def _acquire():
            with func(*args, **kwargs) as f:
                yield f
        acquire_gen = _acquire()
        def release_func():
            try:
                next(acquire_gen)
            except StopIteration:
                pass
        return acquire_gen, release_func
    def __init__(self, func, *args, **kwargs):
        self._acq, self._rel = self.__class__.enter_context(func, *args, **kwargs)
    def acquire(self):
        next(self._acq)
        return True
    def release(self):
        self._rel()
    # Traditional Dijkstra names.
    P = acquire
    V = release

然后，我可以围绕 third.party.lock 创建包装器，如下所示...

mylocker = ContextWrapper(third.party.lock, arg0, arg1)

... 我可以调用 acquire 和 release 如下 ...

mylocker.acquire()
# or mylocker.P()
mylocker.release()
# or mylocker.V()

这是另一个如何使用此 class ...

的示例

class ThirdPartyLockClass(ContextWrapper):
    def __init__(self, arg0, arg1):
        # do any initialization
        super().__init__(third.party.lock, arg0, arg1)
    # Implementing the following methods is optional.
    # This is only needed if this class intends to do more
    # than the wrapped class during `acquire` or `release`,
    # such as logging, etc.
    def acquire(self):
        # do whatever
        rc = super().acquire()
        # maybe to other stuff
        return rc
    def release(self):
        # do whatever
        super().release()
        # maybe do other stuff
    P = acquire
    V = release

mylocker = ThirdPartyLockClass(arg0, arg1)
mylocker.acquire()
# or mylocker.P()
mylocker.release()
# or mylocker.V()

... 或者我什至可以在没有向第三方锁添加额外功能的简单情况下执行以下操作 class ...

class GenericLocker(ContextWrapper):
    def __init__(self, func, *args, **kwargs):
        super().__init__(func, *args, **kwargs)

mylocker = GenericLocker(third.party.lock, arg0, arg1)
mylocker.acquire()
# or mylocker.P()
mylocker.release()
# or mylocker.V()

Answer 2

为什么不显式调用 contextlib 包装器的 __enter__ 和 __exit__？

import contextlib

# Something to wrap
class Test:
    def __init__(self):
        print('Opening Test')
    def use(self):
        print('Using Test')
    def close(self):
        print('Closing Test')

# Wrap in the standard contextlib
@contextlib.contextmanager
def test():
    try:
        t = Test()
        yield t
    finally:
        t.close()

# Does it work?  yes...
with test() as t:
    t.use()

# Implementing acquire/release...
class TestWrapper(object):
    def __init__(self):
        pass
    def acquire(self):
        self.t = test()
        return self.t.__enter__()
    def release(self):
        self.t.__exit__(None,None,None)

# Also works...
tw = TestWrapper()
t = tw.acquire()
t.use()
tw.release()

输出：

Opening Test
Using Test
Closing Test
Opening Test
Using Test
Closing Test

python3/contextlib: 将@contextlib.contextmanager 转换为 acquire/release？

python3/contextlib: Converting @contextlib.contextmanager to acquire/release?

generator

decorator

contextmanager

python-3.x