Python 多个上下文管理器合二为一 class

Python multiple context managers in one class

我希望能够编写这样的代码:

with obj.in_batch_mode:
    obj.some_attr = "some_value"
    obj.some_int = 142
    ...

当我希望 obj 等待发送有关自身的更新,直到多个作业完成。我在 __setattr__ 上有钩子,需要一些时间才能 运行,更改可以一起发送。

我不想使用这样的代码,因为它增加了忘记离开 batch_mode 的风险(这正是 with 关键字的好处):

obj.enter_batch_mode()
obj.some_attr = "some_value"
obj.some_int = 142
...
obj.exit_batch_mode()

我还没弄清楚如何实现它。只需键入 with obj:(并简单地在 obj 上实现 with)并不能像描述性的那样阅读任何地方。

也许是这样的:

实施助手class

class WithHelperObj(object):
    def __init__(self,obj):
        self.obj = obj

    def __enter__(self):
        self.obj.impl_enter_batch()

    def __exit__(self, exc_type, exc_value, traceback):
        self.obj.impl_exit_batch()

    class MyObject(object):
        def in_batch_mode(self):
            return WithHelperObj(self)

在 class 本身中,实现方法而不是字段,与 with 语句一起使用

    def impl_enter_batch(self):
        print 'In impl_enter_batch'

    def impl_exit_batch(self):
        print 'In impl_exit_batch'

    def doing(self):
        print 'doing'

然后使用它:

o = MyObject()
with o.in_batch_mode():
    o.doing()

如果您寻求简单的解决方案并且不需要任何嵌套模式更改(例如,从 STD 到 BATCH 到 VERBOSE 回到 BATCH 回到 STD)

class A(object):
    STD_MODE = 'std' 
    BATCH_MODE = 'batch'
    VERBOSE_MODE = 'verb'

    def __init__(self):
        self.mode = self.STD_MODE

    def in_mode(self, mode):
        self.mode = mode
        return self

    def __enter__(self):
        return self

    def __exit__(self, type, value, tb):
        self.mode = self.STD_MODE

obj = A()
print obj.mode
with obj.in_mode(obj.BATCH_MODE) as x:
    print x.mode
print obj.mode

产出

std
batch
std

这建立在 Pynchia 的回答之上,但增加了对多种模式的支持并允许嵌套 with 语句,即使多次使用相同的模式也是如此。它缩放 O(#nested_modes),基本上是 O(1)

只记得使用堆栈来存储与模式相关的数据。

class A():
    _batch_mode = "batch_mode"
    _mode_stack = []

    @property
    def in_batch_mode(self):
        self._mode_stack.append(self._batch_mode)
        return self

    def __enter__(self):
        return self

    def __exit__(self, type, value, tb):
        self._mode_stack.pop()
        if self._batch_mode not in self._mode_stack:
            self.apply_edits()

然后我在任何需要的地方都有这些支票:

if self._batch_mode not in self._mode_stack:
    self.apply_edits()

也可以使用模式方法:

with x.in_some_mode(my_arg):

请记住将 my_arg 保存在 x 中的堆栈中,并在从模式堆栈中弹出该模式时将其从堆栈中清除。


使用该对象的代码现在可以

with obj.in_batch_mode:
    obj.some_property = "some_value"

并且嵌套没有问题,因此我们可以在任何地方添加另一个 with obj.in_some_mode:,而不会出现任何难以调试的错误,也不必检查调用的每个函数以确保对象的 with-语句从不嵌套:

def b(obj):
    with obj.in_batch_mode:
        obj.some_property = "some_value"

x = A()
with x.in_batch_mode:
    x.my_property = "my_value"
    b(x)

通常,实现上下文管理器的一种非常简单的方法是使用 contextlib 模块。编写上下文管理器变得和编写单个 yield 生成器一样简单。在yield替换__enter__方法之前,yield的对象是__enter__的return值,yield之后的部分是__exit__方法。 class 上的任何函数都可以是上下文管理器,它只需要这样装饰即可。例如,采用这个简单的 ConsoleWriter class:

from contextlib import contextmanager

from sys import stdout
from io import StringIO
from functools import partial

class ConsoleWriter:

    def __init__(self, out=stdout, fmt=None):
        self._out = out
        self._fmt = fmt

    @property
    @contextmanager
    def batch(self):
        original_out = self._out
        self._out = StringIO()
        try:
            yield self
        except Exception as e:
            # There was a problem. Ignore batch commands.
            # (do not swallow the exception though)
            raise
        else:
            # no problem
            original_out.write(self._out.getvalue())
        finally:
            self._out = original_out

    @contextmanager
    def verbose(self, fmt="VERBOSE: {!r}"):
        original_fmt = self._fmt
        self._fmt = fmt
        try:
            yield self
        finally:
            # don't care about errors, just restore end
            self._fmt = original_fmt

    def __getattr__(self, attr):
        """creates function that writes capitalised attribute three times"""
        return partial(self.write, attr.upper()*3)


    def write(self, arg):
        if self._fmt:
            arg = self._fmt.format(arg)
        print(arg, file=self._out)

用法示例:

writer = ConsoleWriter()
with writer.batch:
    print("begin batch")
    writer.a()
    writer.b()
    with writer.verbose():
        writer.c()
    print("before reentrant block")
    with writer.batch:
        writer.d()
    print("after reentrant block")
    print("end batch -- all data is now flushed")

输出:

begin batch
before reentrant block
after reentrant block
end batch -- all data is now flushed
AAA
BBB
VERBOSE: 'CCC'
DDD