contextlib —具有语句环境的 Util

源代码： Lib/contextlib.py

---

该模块为涉及with语句的常见任务提供 Util。有关更多信息，另请参见上下文 Management 器类型和使用语句上下文 Management 器。

Utilities

提供的函数和类：

• 类别 contextlib. AbstractContextManager
  • 实现抽象 Base Class和object.exit()的类的抽象 Base Class。提供了object.enter()的默认实现，该实现返回self，而object.exit()是抽象方法，默认情况下返回None。另请参见上下文 Management 器类型的定义。

3.6 版的新Function。

• 类别 contextlib. AbstractAsyncContextManager
  • 实现抽象 Base Class和object.aexit()的类的抽象 Base Class。提供了object.aenter()的默认实现，该实现返回self，而object.aexit()是抽象方法，默认情况下返回None。另请参见异步上下文 Management 器的定义。

3.7 版中的新Function。

• @ contextlib. contextmanager
  • 此函数是decorator，可用于为with语句上下文 Management 器定义工厂函数，而无需创建类或单独的enter()和exit()方法。

尽管许多对象本机支持在 with 语句中使用，但有时需要 Management 的资源本身不是上下文 Management 器，并且没有实现与contextlib.closing一起使用的close()方法

下面是一个抽象的示例，以确保正确的资源 Management：

from contextlib import contextmanager

@contextmanager
def managed_resource(*args, **kwds):
    # Code to acquire resource, e.g.:
    resource = acquire_resource(*args, **kwds)
    try:
        yield resource
    finally:
        # Code to release resource, e.g.:
        release_resource(resource)

>>> with managed_resource(timeout=3600) as resource:
...     # Resource is released at the end of this block,
...     # even if code in the block raises an exception

被修饰的函数在调用时必须返回generator -iterator。此迭代器必须精确地产生一个值，该值将绑定到with语句的as子句中的目标(如果有)。

在生成器屈服的 Moment，执行嵌套在with语句中的块。退出该块后，将恢复生成器。如果在块中发生未处理的异常，则会在生成器内部产生yield的位置将其重新引发。因此，您可以使用try…except…finally语句来捕获错误(如果有)，或确保进行一些清除。如果仅出于记录日志或执行某些操作(而不是完全抑制它)的目的捕获了异常，则生成器必须重新引发该异常。否则，生成器上下文 Management 器将向with语句指示已处理了该异常，并且将在with语句之后立即使用该语句 continue 执行。

contextmanager()使用ContextDecorator，因此它创建的上下文 Management 器可以用作修饰符，也可以用作with语句。当用作装饰器时，将在每个函数调用上隐式创建一个新的生成器实例(这使contextmanager()创建的否则“一次性”上下文 Management 器可以满足上下文 Management 器支持多个调用以便用作装饰器的要求) 。

在版本 3.2 中更改：使用ContextDecorator。

• @ contextlib. asynccontextmanager
  • 与contextmanager()类似，但创建了异步上下文 Management 器。

此函数是decorator，可用于为async with语句异步上下文 Management 器定义工厂函数，而无需创建类或单独的aenter()和aexit()方法。它必须应用于asynchronous generator函数。

一个简单的例子：

from contextlib import asynccontextmanager

@asynccontextmanager
async def get_connection():
    conn = await acquire_db_connection()
    try:
        yield conn
    finally:
        await release_db_connection(conn)

async def get_all_users():
    async with get_connection() as conn:
        return conn.query('SELECT ...')

3.7 版中的新Function。

• contextlib. closing(* thing *)
  • 返回一个上下文 Management 器，该 Management 器在块完成时关闭* thing *。这基本上等效于：

from contextlib import contextmanager

@contextmanager
def closing(thing):
    try:
        yield thing
    finally:
        thing.close()

并让您编写如下代码：

from contextlib import closing
from urllib.request import urlopen

with closing(urlopen('http://www.python.org')) as page:
    for line in page:
        print(line)

无需显式关闭page。即使发生错误，退出with块时也会调用page.close()。

• contextlib. nullcontext(* enter_result = None *)
  • 返回一个上下文 Management 器，该上下文 Management 器从__enter__返回* enter_result *，否则不执行任何操作。它旨在用作可选上下文 Management 器的替代，例如：

def myfunction(arg, ignore_exceptions=False):
    if ignore_exceptions:
        # Use suppress to ignore all exceptions.
        cm = contextlib.suppress(Exception)
    else:
        # Do not ignore any exceptions, cm has no effect.
        cm = contextlib.nullcontext()
    with cm:
        # Do something

使用* enter_result *的示例：

def process_file(file_or_path):
    if isinstance(file_or_path, str):
        # If string, open file
        cm = open(file_or_path)
    else:
        # Caller is responsible for closing file
        cm = nullcontext(file_or_path)

    with cm as file:
        # Perform processing on the file

3.7 版中的新Function。

• contextlib. suppress(*exception)
  • 返回一个上下文 Management 器，该上下文 Management 器将抑制任何指定的异常(如果它们出现在 with 语句的主体中)，然后在 with 语句结束后从第一个语句恢复执行。

与任何其他完全抑制异常的机制一样，此上下文 Management 器仅应用于涵盖非常具体的错误，在这些错误中，静默地 continue 执行程序是正确的做法。

For example:

from contextlib import suppress

with suppress(FileNotFoundError):
    os.remove('somefile.tmp')

with suppress(FileNotFoundError):
    os.remove('someotherfile.tmp')

此代码等效于：

try:
    os.remove('somefile.tmp')
except FileNotFoundError:
    pass

try:
    os.remove('someotherfile.tmp')
except FileNotFoundError:
    pass

该上下文 Management 器是reentrant。

3.4 版的新Function。

• contextlib. redirect_stdout(* new_target *)
  • 上下文 Management 器，用于临时将sys.stdout重定向到另一个文件或类似文件的对象。

该工具为现有Function或类的输出增加了灵 Active，这些Function或类的输出被硬连线到 stdout。

例如，help()的输出通常发送到* sys.stdout *。您可以pass将输出重定向到io.StringIO对象来捕获该输出的字符串：

f = io.StringIO()
with redirect_stdout(f):
    help(pow)
s = f.getvalue()

要将help()的输出发送到磁盘上的文件，请将输出重定向到常规文件：

with open('help.txt', 'w') as f:
    with redirect_stdout(f):
        help(pow)

要将help()的输出发送到* sys.stderr *：

with redirect_stdout(sys.stderr):
    help(pow)

请注意，对sys.stdout的全局副作用意味着此上下文 Management 器不适合在库代码和大多数线程应用程序中使用。它还对子流程的输出没有影响。但是，对于许多 Util 脚本，它仍然是一种有用的方法。

该上下文 Management 器是reentrant。

3.4 版的新Function。

• contextlib. redirect_stderr(* new_target *)
  • 与redirect_stdout()类似，但将sys.stderr重定向到另一个文件或类似文件的对象。

该上下文 Management 器是reentrant。

3.5 版中的新Function。

• 类别 contextlib. ContextDecorator
  • 使上下文 Management 器也可以用作装饰器的 Base Class。

从ContextDecorator继承的上下文 Management 器必须照常实现__enter__和__exit__。 __exit__保留其可选的异常处理，即使用作装饰器也是如此。

ContextDecorator由contextmanager()使用，因此您会自动获得此Function。

ContextDecorator的示例：

from contextlib import ContextDecorator

class mycontext(ContextDecorator):
    def __enter__(self):
        print('Starting')
        return self

    def __exit__(self, *exc):
        print('Finishing')
        return False

>>> @mycontext()
... def function():
...     print('The bit in the middle')
...
>>> function()
Starting
The bit in the middle
Finishing

>>> with mycontext():
...     print('The bit in the middle')
...
Starting
The bit in the middle
Finishing

对于以下任何形式的构造，此更改仅是语法糖：

def f():
    with cm():
        # Do stuff

ContextDecorator可让您写：

@cm()
def f():
    # Do stuff

很明显，cm不仅适用于整个Function，还适用于整个Function(保存缩进级别也很不错)。

可以pass使用ContextDecorator作为 mixin 类来扩展已经具有 Base Class 的现有上下文 Management 器：

from contextlib import ContextDecorator

class mycontext(ContextBaseClass, ContextDecorator):
    def __enter__(self):
        return self

    def __exit__(self, *exc):
        return False

3.2 版中的新Function。

• 类别 contextlib. ExitStack
  • 一个上下文 Management 器，旨在使其以编程方式轻松组合其他上下文 Management 器和清除Function，尤其是那些可选的或由 Importing 数据驱动的清除Function。

例如，一组文件可以很容易地在一个 with 语句中进行如下处理：

with ExitStack() as stack:
    files = [stack.enter_context(open(fname)) for fname in filenames]
    # All opened files will automatically be closed at the end of
    # the with statement, even if attempts to open files later
    # in the list raise an exception

每个实例维护一堆注册的回调，当实例关闭时(在with语句的末尾显式或隐式)，将以相反的 Sequences 调用这些回调。注意，当上下文堆栈实例被垃圾回收时，回调不会被隐式调用。

使用此堆栈模型，以便可以正确处理以__init__方法获取其资源的上下文 Management 器(例如文件对象)。

由于已注册的回调是按照注册的相反 Sequences 调用的，因此finally行为就好像已将多个嵌套的with语句与已注册的一组回调一起使用。这甚至扩展到异常处理-如果内部回调抑制或替换异常，则将根据该更新状态将外部回调传递给参数。

这是一个相对较低级别的 API，它负责正确展开退出回调堆栈的细节。它为以特定于应用程序的方式操纵退出堆栈的高级上下文 Management 器提供了合适的基础。

版本 3.3 中的新Function。

• enter_context(* cm *)
  • Importing 一个新的上下文 Management 器，并将其exit()方法添加到回调堆栈中。返回值是上下文 Management 器自己的enter()方法的结果。

这些上下文 Management 器可以像直接将它们用作with语句的一部分一样抑制异常。

• push(退出)
  • 将上下文 Management 器的exit()方法添加到回调堆栈。

由于未调用__enter__，因此此方法可用于使用上下文 Management 器自己的exit()方法覆盖enter()实现的一部分。

如果传递的对象不是上下文 Management 器，则此方法假定它是一个具有与上下文 Management 器的exit()方法相同签名的回调，并将其直接添加到回调堆栈中。

pass返回真值，这些回调可以像上下文 Management 器exit()方法一样抑制异常。

传入的对象从函数返回，从而允许将此方法用作函数装饰器。

• callback((* callback *， *args ， * kwds *)
  • 接受任意的回调函数和参数，并将其添加到回调堆栈中。

与其他方法不同，以这种方式添加的回调无法抑制异常(因为它们从未传递过异常详细信息)。

传入的回调从函数返回，从而允许将此方法用作函数装饰器。

• pop_all ( )
  • 将回调堆栈转移到一个新的ExitStack实例并返回它。此操作不会调用任何回调-而是现在在关闭新堆栈时(在with语句的末尾显式或隐式)将调用它们。

例如，可以按“全部或全部”操作打开一组文件，如下所示：

with ExitStack() as stack:
    files = [stack.enter_context(open(fname)) for fname in filenames]
    # Hold onto the close method, but don't call it yet.
    close_files = stack.pop_all().close
    # If opening any file fails, all previously opened files will be
    # closed automatically. If all files are opened successfully,
    # they will remain open even after the with statement ends.
    # close_files() can then be invoked explicitly to close them all.

• close ( )

  • 立即展开回调堆栈，以与注册相反的 Sequences 调用回调。对于任何已注册的上下文 Management 器和退出回调，传入的参数将指示未发生异常。

• 类别 contextlib. AsyncExitStack

  • 异步上下文 Management 器与ExitStack类似，它支持将同步上下文 Management 器和异步上下文 Management 器组合在一起，并且具有用于清理逻辑的协程。

尚未实现close()方法，而必须使用aclose()。

• enter_async_context(* cm *)

  • 与enter_context()类似，但需要异步上下文 Management 器。

• push_async_exit(退出)

  • 与push()类似，但是需要异步上下文 Management 器或协程函数。

• push_async_callback((* callback *， *args ， * kwds *)

  • 与callback()类似，但需要协程Function。

• aclose ( )

  • 与close()类似，但可以正确处理 await 项。

continueasynccontextmanager()的示例：

async with AsyncExitStack() as stack:
    connections = [await stack.enter_async_context(get_connection())
        for i in range(5)]
    # All opened connections will automatically be released at the end of
    # the async with statement, even if attempts to open a connection
    # later in the list raise an exception.

3.7 版中的新Function。

示例和食谱

本节描述了一些示例和秘诀，以有效利用contextlib提供的工具。

支持可变数量的上下文 Management 器

ExitStack的主要用例是类文档中给出的用例：在单个with语句中支持可变数量的上下文 Management 器和其他清除操作。可变性可能来自于需要由用户 Importing 驱动的上下文 Management 器的数量(例如，打开用户指定的文件集合)，或者来自某些可选的上下文 Management 器：

with ExitStack() as stack:
    for resource in resources:
        stack.enter_context(resource)
    if need_special_resource():
        special = acquire_special_resource()
        stack.callback(release_special_resource, special)
    # Perform operations that use the acquired resources

如图所示，ExitStack还使使用with语句 Management 本来不支持上下文 Management 协议的任意资源变得非常容易。

从__enter_方法捕获异常

偶尔希望从__enter__方法实现中捕获异常，而不会无意中从with语句主体或上下文 Management 器的__exit__方法中捕获异常。pass使用ExitStack，可以稍微分开上下文 Management 协议中的步骤，以实现以下目的：

stack = ExitStack()
try:
    x = stack.enter_context(cm)
except Exception:
    # handle __enter__ exception
else:
    with stack:
        # Handle normal case

实际需要执行的操作很可能表明底层 API 应该提供直接的资源 Management 接口，以与try/except/finally语句一起使用，但是并非所有 API 在这方面都经过精心设计。如果上下文 Management 器是唯一提供的资源 ManagementAPI，则ExitStack可以使处理在with语句中无法直接处理的各种情况变得更加容易。

__enter_实现中的清理

如ExitStack.push()的文档所述，如果enter()实现中的后续步骤失败，则此方法可用于清理已分配的资源。

这是一个为上下文 Management 器执行此操作的示例，该 Management 器接受资源获取和释放Function以及可选的验证Function，并将它们 Map 到上下文 Management 协议：

from contextlib import contextmanager, AbstractContextManager, ExitStack

class ResourceManager(AbstractContextManager):

    def __init__(self, acquire_resource, release_resource, check_resource_ok=None):
        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

    @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__(self):
        resource = self.acquire_resource()
        with self._cleanup_on_error():
            if not self.check_resource_ok(resource):
                msg = "Failed validation for {!r}"
                raise RuntimeError(msg.format(resource))
        return resource

    def __exit__(self, *exc_details):
        # We don't need to duplicate any of our resource release logic
        self.release_resource()

替换使用 try-finally 和 flag 变量

您有时会看到的模式是带有标志变量的try-finally语句，该语句指示是否应执行finally子句的主体。以最简单的形式(不能仅仅pass使用except子句来处理)，它看起来像这样：

cleanup_needed = True
try:
    result = perform_operation()
    if result:
        cleanup_needed = False
finally:
    if cleanup_needed:
        cleanup_resources()

与任何基于try语句的代码一样，这可能会导致开发和审查方面的问题，因为设置代码和清除代码finally可能会被任意长的代码段分隔开。

ExitStack使得可以替代地在with语句的末尾注册要执行的回调，然后稍后决定跳过执行该回调：

from contextlib import ExitStack

with ExitStack() as stack:
    stack.callback(cleanup_resources)
    result = perform_operation()
    if result:
        stack.pop_all()

这允许预先明确地进行预期的清理行为，而不需要单独的标志变量。

如果特定的应用程序经常使用此模式，则可以pass一个小的帮助程序类进一步简化该模式：

from contextlib import ExitStack

class Callback(ExitStack):
    def __init__(self, callback, /, *args, **kwds):
        super(Callback, self).__init__()
        self.callback(callback, *args, **kwds)

    def cancel(self):
        self.pop_all()

with Callback(cleanup_resources) as cb:
    result = perform_operation()
    if result:
        cb.cancel()

如果资源清理尚未很好地 Binding 到独立函数中，那么仍然可以使用ExitStack.callback()的修饰符形式提前语句资源清理：

from contextlib import ExitStack

with ExitStack() as stack:
    @stack.callback
    def cleanup_resources():
        ...
    result = perform_operation()
    if result:
        stack.pop_all()

由于装饰器协议的工作方式，以这种方式语句的回调函数不能使用任何参数。相反，任何要释放的资源都必须作为闭包变量进行访问。

使用上下文 Management 器作为函数装饰器

ContextDecorator使得可以在普通的with语句中以及作为函数装饰器使用上下文 Management 器。

例如，有时使用可以跟踪进入时间和退出时间的 Logger 包装函数或语句组很有用。从ContextDecorator继承不提供任务的函数装饰器和上下文 Management 器，而是在单个定义中提供了这两种Function：

from contextlib import ContextDecorator
import logging

logging.basicConfig(level=logging.INFO)

class track_entry_and_exit(ContextDecorator):
    def __init__(self, name):
        self.name = name

    def __enter__(self):
        logging.info('Entering: %s', self.name)

    def __exit__(self, exc_type, exc, exc_tb):
        logging.info('Exiting: %s', self.name)

此类的实例可以用作上下文 Management 器：

with track_entry_and_exit('widget loader'):
    print('Some time consuming activity goes here')
    load_widget()

并作为Function装饰器：

@track_entry_and_exit('widget loader')
def activity():
    print('Some time consuming activity goes here')
    load_widget()

请注意，将上下文 Management 器用作函数装饰器时，还有一个附加限制：无法访问enter()的返回值。如果需要该值，则仍然必须使用显式的with语句。

一次性使用，可重用和可重入的上下文 Management 器

大多数上下文 Management 器的编写方式意味着它们只能在with语句中有效使用一次。这些一次性使用的上下文 Management 器必须在每次使用时都重新创建-try再次使用它们会触发异常，否则将无法正常工作。

这个共同的限制意味着通常建议直接在使用它们的with语句的标题中直接创建上下文 Management 器(如以上所有用法示例所示)。

文件是有效的单次使用上下文 Management 器的一个示例，因为第一个with语句将关闭文件，从而阻止使用该文件对象进行任何进一步的 IO 操作。

使用contextmanager()创建的上下文 Management 器也是一次性使用的上下文 Management 器，如果再次try使用它们，则会抱怨底层生成器无法屈服：

>>> from contextlib import contextmanager
>>> @contextmanager
... def singleuse():
...     print("Before")
...     yield
...     print("After")
...
>>> cm = singleuse()
>>> with cm:
...     pass
...
Before
After
>>> with cm:
...     pass
...
Traceback (most recent call last):
    ...
RuntimeError: generator didn't yield

可重入上下文 Management 器

更复杂的上下文 Management 器可能是“可重入的”。这些上下文 Management 器不仅可以在多个with语句中使用，还可以在已经使用同一上下文 Management 器的with语句内部使用。

threading.RLock是可重入上下文 Management 器的一个示例，suppress()和redirect_stdout()也是。这是一个非常简单的可重用示例：

>>> from contextlib import redirect_stdout
>>> from io import StringIO
>>> stream = StringIO()
>>> write_to_stream = redirect_stdout(stream)
>>> with write_to_stream:
...     print("This is written to the stream rather than stdout")
...     with write_to_stream:
...         print("This is also written to the stream")
...
>>> print("This is written directly to stdout")
This is written directly to stdout
>>> print(stream.getvalue())
This is written to the stream rather than stdout
This is also written to the stream

现实世界中的重入示例更可能涉及多个相互调用的函数，因此比该示例要复杂得多。

还要注意，可重入与线程安全不是同一件事。例如，redirect_stdout()绝对不是线程安全的，因为它pass将sys.stdout绑定到其他流来对系统状态进行全局修改。

可重用的上下文 Management 器

与一次性使用和可重入上下文 Management 器截然不同的是“可重用”上下文 Management 器(或者，完全明确的是，“可重用但不可重入”上下文 Management 器，因为可重入上下文 Management 器也可重用)。这些上下文 Management 器支持多次使用，但是如果特定的上下文 Management 器实例已在包含 with 语句中使用，则将失败(否则将无法正常工作)。

threading.Lock是可重用但不是可重入的上下文 Management 器的示例(对于可重入锁，必须使用threading.RLock代替)。

可重用但不能重入的上下文 Management 器的另一个示例是ExitStack，因为它在离开任何 with 语句时会调用* all *当前注册的回调，而不管这些回调的添加位置如何：

>>> from contextlib import ExitStack
>>> stack = ExitStack()
>>> with stack:
...     stack.callback(print, "Callback: from first context")
...     print("Leaving first context")
...
Leaving first context
Callback: from first context
>>> with stack:
...     stack.callback(print, "Callback: from second context")
...     print("Leaving second context")
...
Leaving second context
Callback: from second context
>>> with stack:
...     stack.callback(print, "Callback: from outer context")
...     with stack:
...         stack.callback(print, "Callback: from inner context")
...         print("Leaving inner context")
...     print("Leaving outer context")
...
Leaving inner context
Callback: from inner context
Callback: from outer context
Leaving outer context

如该示例的输出所示，跨多个 with 语句重用单个堆栈对象可以正常工作，但是try嵌套它们将导致堆栈在最里面的 with 语句末尾被清除，这不太可能是理想的行为。

使用单独的ExitStack实例而不是重用单个实例可以避免该问题：

>>> from contextlib import ExitStack
>>> with ExitStack() as outer_stack:
...     outer_stack.callback(print, "Callback: from outer context")
...     with ExitStack() as inner_stack:
...         inner_stack.callback(print, "Callback: from inner context")
...         print("Leaving inner context")
...     print("Leaving outer context")
...
Leaving inner context
Callback: from inner context
Leaving outer context
Callback: from outer context