Pythonic方法来为类所拥有的对象组合上下文管理器

45

通常需要多个具有资源的对象明确释放,例如两个文件; 当任务局限在一个函数内时,可以使用嵌套的 with 块轻松实现这一点,或者更好的是,使用一个带有多个 with_item 子句的单个 with 块:

with open('in.txt', 'r') as i, open('out.txt', 'w') as o:
    # do stuff

然而,我仍然很难理解当这些对象不仅仅是函数作用域内的本地对象,而是由类实例所拥有时,它们应该如何工作 - 换句话说,上下文管理器是如何组合的。

理想情况下,我想做这样的事情:

class Foo:
    def __init__(self, in_file_name, out_file_name):
        self.i = WITH(open(in_file_name, 'r'))
        self.o = WITH(open(out_file_name, 'w'))

并使Foo本身成为一个上下文管理器来处理io,这样当我执行以下操作时:

with Foo('in.txt', 'out.txt') as f:
    # do stuff

self.iself.o会被自动处理,就像你期望的一样。

我曾尝试编写以下内容:

class Foo:
    def __init__(self, in_file_name, out_file_name):
        self.i = open(in_file_name, 'r').__enter__()
        self.o = open(out_file_name, 'w').__enter__()

    def __enter__(self):
        return self

    def __exit__(self, *exc):
        self.i.__exit__(*exc)
        self.o.__exit__(*exc)

但是它既冗长又不安全,容易在构造函数中发生异常。搜索一段时间后,我找到了这篇2015年的博客文章,它使用contextlib.ExitStack获得了与我想要的非常相似的东西:

class Foo(contextlib.ExitStack):
    def __init__(self, in_file_name, out_file_name):
        super().__init__()
        self.in_file_name = in_file_name
        self.out_file_name = out_file_name

    def __enter__(self):
        super().__enter__()
        self.i = self.enter_context(open(self.in_file_name, 'r')
        self.o = self.enter_context(open(self.out_file_name, 'w')
        return self
这相当令人满意,但我对以下事实感到困惑:
  • 在文档中找不到关于此用法的任何信息,所以似乎不是解决此问题的“官方”方法;
  • 一般来说,我发现很难找到有关此问题的信息,这让我想到我正在尝试将非pythonic的解决方案应用于该问题。

一些额外的背景信息:我主要使用C ++,在其中对于此问题不存在块范围案例和对象范围案例之间的区别,因为这种清理是在析构函数(类似__del__,但是以确定性方式调用)内部实现的,并且析构函数(即使未显式定义)会自动调用子对象的析构函数。因此,以下两种情况都可以:
{
    std::ifstream i("in.txt");
    std::ofstream o("out.txt");
    // do stuff
}

struct Foo {
    std::ifstream i;
    std::ofstream o;

    Foo(const char *in_file_name, const char *out_file_name) 
        : i(in_file_name), o(out_file_name) {}
}

{
    Foo f("in.txt", "out.txt");
}

让所有的清理工作自动完成,就像你通常希望的那样。

我正在寻找Python中类似的解决方案,但我担心我只是试图应用来自C ++的模式,而根本问题有一个完全不同的解决方案,我想不出来。


因此,总结一下:对于拥有需要清理的对象的对象成为上下文管理器本身的问题,有什么Pythonic的解决方案,可以正确调用其子项的 __enter__ / __exit__ ?


4
我认为使用ExitStack解决方案非常符合Python编程风格。 - BrenBarn
@BrenBarn:很高兴知道这个,但我仍然有点担心,因为在一个随机的博客中只提到了一次这个解决方案,而不是在官方文档中,对于我认为应该是一个相当常见的问题。这让我感到困惑。 - Matteo Italia
3
我不确定为什么你期望这个内容出现在官方文档中。通常,官方文档只记录事物的工作原理,而不是它们的用途。有很多常见问题的解决方法在官方文档中没有解释。这里是一个相关的问题,Martijn Pieters在他的回答评论中建议为类似目的使用ExitStack的子类。 - BrenBarn
1
一个上下文管理器,旨在使编程者轻松地组合其他上下文管理器和清理函数,特别是那些可选的或由输入数据驱动的函数。我觉得文档表明ExitStack解决方案非常符合Pythonic风格。 - Edward Minnix
5个回答

18
我认为contextlib.ExitStack是Pythonic和规范的解决方案,非常适合这个问题。本回答的其余部分旨在展示我得出这个结论的链接和思路:

Python的原始增强请求

https://bugs.python.org/issue13585

最初的想法+实现是作为Python标准库增强提出的,附有推理和示例代码。它得到了核心开发人员Raymond Hettinger和Eric Snow等人的详细讨论。该问题的讨论清楚地展示了最初想法的发展,成为适用于标准库并符合Python风格的东西。对该主题串的尝试总结如下:

nikratio最初提出:

我想建议将CleanupManager类添加到contextlib模块中,该类在http://article.gmane.org/gmane.comp.python.ideas/12447中有描述。这个想法是添加一个通用的上下文管理器来管理(Python或非Python)资源,这些资源没有自己的上下文管理器

遭到rhettinger的关注:

到目前为止,这方面还没有任何需求,我也没有看到像它这样的代码被用于实际环境中。据我所知,它并没有明显优于一个简单的try/finally块。

对此进行了长时间的讨论,ncoghlan发表了如下言论:

TestCase.setUp()和TestCase.tearDown()是__enter__()和exit()的前身之一。addCleanUp()在这里扮演完全相同的角色,我已经看到许多正面反馈向Michael表示感谢unittest API的这个补充。... ...在这种情况下,自定义上下文管理器通常不是一个好主意,因为它们使可读性变得更差(依赖于人们理解上下文管理器的作用)。另一方面,基于标准库的解决方案提供了最佳选择: - 代码变得更容易正确编写和审计(出于添加with语句的原始原因) - 这个习惯用法最终会为所有Python用户所熟悉... ...如果你愿意,我可以在python-dev上接手这个问题,但我希望说服你,有这样的需求...

稍后ncoghlan再次提到:

我之前的描述并不完整 - 一旦我开始组合contextlib2,这个CleanupManager的想法很快演变成了ContextStack [1],这是一个更强大的工具,可以以不必与源代码中的词汇作用域相对应的方式来操作上下文管理器。

ExitStack 的示例/配方/博客文章 标准库源代码中有几个示例和配方,您可以在添加此功能的合并修订中查看:https://hg.python.org/cpython/rev/8ef66c73b1e1

此外,原问题创建者(Nikolaus Rath / nikratio)撰写了一篇博客文章,生动地描述了为什么ContextStack是一种好的模式,并提供了一些使用示例:https://www.rath.org/on-the-beauty-of-pythons-exitstack.html


10
你的第二个例子是在Python中最直接和符合Pythonic方式去实现它的。然而,你的例子仍然有一个错误。如果第二个open()期间引发了异常,
self.i = self.enter_context(open(self.in_file_name, 'r')
self.o = self.enter_context(open(self.out_file_name, 'w') # <<< HERE

那么,如果Foo.__enter__()未能成功返回,self.i将不会在您期望的时间被释放,因为Foo.__exit__()不会被调用。要解决此问题,请在每个上下文调用中使用try-except语句,在发生异常时调用Foo.__exit__()

import contextlib
import sys

class Foo(contextlib.ExitStack):

    def __init__(self, in_file_name, out_file_name):
        super().__init__()
        self.in_file_name = in_file_name
        self.out_file_name = out_file_name

    def __enter__(self):
        super().__enter__()

        try:
            # Initialize sub-context objects that could raise exceptions here.
            self.i = self.enter_context(open(self.in_file_name, 'r'))
            self.o = self.enter_context(open(self.out_file_name, 'w'))

        except:
            if not self.__exit__(*sys.exc_info()):
                raise

        return self

6
我认为使用助手更好:
from contextlib import ExitStack, contextmanager

class Foo:
    def __init__(self, i, o):
        self.i = i
        self.o = o

@contextmanager
def multiopen(i, o):
    with ExitStack() as stack:
        i = stack.enter_context(open(i))
        o = stack.enter_context(open(o))
        yield Foo(i, o)

使用方式与原生的open方法类似:

with multiopen(i_name, o_name) as foo:
    pass

你的第一个代码片段和我写的那个一样有漏洞,当我尝试编写一些东西时(如果在执行第二个“open”时出现异常怎么办?);另外,为什么我要记得关闭()这些东西?这只是样板代码,可能会写错;上下文管理器(以及C++中的析构函数)的目的是让这些自动处理,无论我们如何退出上下文。 - Matteo Italia
1
我之前没有考虑过这个问题,现在正在思考。但是你使用 contextlib.ExitStack 的例子仍然不是一个好主意,因为在这种情况下,用户必须使用 with 语句或者得到的实例具有完全不同的行为。对于 close() 的事情,你的观点也很有道理。但是,“Python 之禅”说:“显式优于隐式。” 对于我来说,我会明确地关闭底层对象,以便我完全知道自己在做什么。当然,使用 __enter__/__exit__ 对也可以。 - Sraw
关于显式优于隐式,这取决于具体情况。资源清理是其中一个领域,在这个领域中,自动化程度越高越好,因为开发人员在这方面通常做得很差,特别是在有异常的语言中。Python拥有自动内存管理,即使“显式优于隐式”,也不是偶然的。 - Matteo Italia
是的,这取决于情况。关于你的例子,我写了另一个例子,我认为它更接近本地的open函数并且更简单。 - Sraw
我注意到你的示例中仍然存在一个bug。当用户手动创建实例时,没有地方调用__exit__,所以如果第二个open失败,它仍然会泄漏。而当使用with语句时,__enter__将被调用两次。在这个简单的示例中没问题,但在真正的任务中可能会引起潜在问题。 - Sraw
显示剩余3条评论

6

正如@cpburnz所提到的,您最后的示例是最好的,但如果第二个打开失败,则会包含错误。避免此错误在标准库文档中有描述。我们可以轻松地从ExitStack documentationResourceManager的示例29.6.2.4 Cleaning up in an __enter__ implementation中调整代码段,以得到MultiResourceManager类:

from contextlib import contextmanager, ExitStack
class MultiResourceManager(ExitStack):
    def __init__(self, resources, acquire_resource, release_resource,
            check_resource_ok=None):
        super().__init__()
        self.acquire_resource = acquire_resource
        self.release_resource = release_resource
        if check_resource_ok is None:
            def check_resource_ok(resource):
                return True
        self.check_resource_ok = check_resource_ok
        self.resources = resources
        self.wrappers = []

    @contextmanager
    def _cleanup_on_error(self):
        with ExitStack() as stack:
            stack.push(self)
            yield
            # The validation check passed and didn't raise an exception
            # Accordingly, we want to keep the resource, and pass it
            # back to our caller
            stack.pop_all()

    def enter_context(self, resource):
        wrapped = super().enter_context(self.acquire_resource(resource))
        if not self.check_resource_ok(wrapped):
            msg = "Failed validation for {!r}"
            raise RuntimeError(msg.format(resource))
        return wrapped

    def __enter__(self):
        with self._cleanup_on_error():
            self.wrappers = [self.enter_context(r) for r in self.resources]
        return self.wrappers

    # NB: ExitStack.__exit__ is already correct

现在你的Foo()类很简单:

import io
class Foo(MultiResourceManager):
    def __init__(self, *paths):
        super().__init__(paths, io.FileIO, io.FileIO.close)

这很好,因为我们不需要任何try-except块 -- 你可能只是使用ContextManagers来首先摆脱它们!然后您可以像您想要的那样使用它(请注意MultiResourceManager.__enter__返回由传递acquire_resource()获取的对象列表):
if __name__ == '__main__':
    open('/tmp/a', 'w').close()
    open('/tmp/b', 'w').close()

    with Foo('/tmp/a', '/tmp/b') as (f1, f2):
        print('opened {0} and {1}'.format(f1.name, f2.name))

我们可以用以下代码片段中的debug_file替换io.FileIO,以便看到其效果:
    class debug_file(io.FileIO):
        def __enter__(self):
            print('{0}: enter'.format(self.name))
            return super().__enter__()
        def __exit__(self, *exc_info):
            print('{0}: exit'.format(self.name))
            return super().__exit__(*exc_info)

然后我们看到:
/tmp/a: enter
/tmp/b: enter
opened /tmp/a and /tmp/b
/tmp/b: exit
/tmp/a: exit

如果我们在循环之前添加import os; os.unlink('/tmp/b'),我们会看到:
/tmp/a: enter
/tmp/a: exit
Traceback (most recent call last):
  File "t.py", line 58, in <module>
    with Foo('/tmp/a', '/tmp/b') as (f1, f2):
  File "t.py", line 46, in __enter__
    self.wrappers = [self.enter_context(r) for r in self.resources]
  File "t.py", line 46, in <listcomp>
    self.wrappers = [self.enter_context(r) for r in self.resources]
  File "t.py", line 38, in enter_context
    wrapped = super().enter_context(self.acquire_resource(resource))
FileNotFoundError: [Errno 2] No such file or directory: '/tmp/b'

您可以看到 /tmp/a 已经被正确地关闭。


4

如果您想为文件处理程序进行处理,最简单的解决方案就是直接将文件处理程序传递给您的类,而不是文件名。

with open(f1, 'r') as f1, open(f2, 'w') as f2:
   with MyClass(f1, f2) as my_obj:
       ...

如果您不需要自定义 __exit__ 功能,甚至可以跳过嵌套的 with 语句。

如果您真的想要将文件名传递给 __init__,则可以像这样解决您的问题:

class MyClass:
     input, output = None, None

     def __init__(self, input, output):
         try:
             self.input = open(input, 'r')
             self.output = open(output, 'w')
         except BaseException as exc:
             self.__exit___(type(exc), exc, exc.__traceback__)
             raise

     def __enter__(self):
         return self

     def __exit__(self, *args):
            self.input and self.input.close()
            self.output and self.output.close()
        # My custom __exit__ code

因此,这真的取决于您的任务,Python有很多选项可供使用。最终,Pythonic的方法是保持您的API简单。


网页内容由stack overflow 提供, 点击上面的
可以查看英文原文,
原文链接