contextlib — with 语句上下文的实用程序 — Python 文档

contextlib — with 语句上下文的实用程序

源代码： :source:`Lib/contextlib.py`

---

该模块为涉及 with 语句的常见任务提供实用程序。 有关更多信息，另请参阅 上下文管理器类型 和 带有语句上下文管理器 。

公用程序

提供的函数和类：

class contextlib.AbstractContextManager

实现 object.__enter__() 和 object.__exit__() 的类的 抽象基类 。 提供了 object.__enter__() 的默认实现，它返回 self 而 object.__exit__() 是一个抽象方法，默认返回 None ]。 另请参阅 上下文管理器类型 的定义。

3.6 版中的新功能。

class contextlib.AbstractAsyncContextManager

实现 object.__aenter__() 和 object.__aexit__() 的类的 抽象基类 。 提供了 object.__aenter__() 的默认实现，它返回 self 而 object.__aexit__() 是一个抽象方法，默认返回 None ]。 另请参阅 异步上下文管理器 的定义。

3.7 版中的新功能。

@contextlib.contextmanager

该函数是一个装饰器，可用于为和语句上下文管理器定义工厂函数，无需创建类或将__enter__()和__exit__() 方法。

虽然许多对象本身支持在 with 语句中使用，但有时需要管理的资源本身不是上下文管理器，并且不实现 contextlib.closing 的 close() 方法]

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

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 子句中的目标，如果有的话。

在生成器产生时，嵌套在 with 语句中的块被执行。 然后在退出块后恢复生成器。 如果块中发生未处理的异常，它会在生成器内在产生的点处重新引发。 因此，您可以使用 try...except...finally 语句来捕获错误（如果有），或确保进行一些清理。 如果一个异常被捕获只是为了记录它或执行某些操作（而不是完全抑制它），则生成器必须重新引发该异常。 否则，生成器上下文管理器将向 with 语句指示异常已被处理，并且将继续执行紧跟在 with 语句之后的语句。

contextmanager() 使用 ContextDecorator 因此它创建的上下文管理器可以用作装饰器以及在 with 语句中。 当用作装饰器时，每个函数调用都会隐式创建一个新的生成器实例（这允许由 contextmanager() 创建的“一次性”上下文管理器满足上下文管理器支持多次调用的要求以便用作装饰器）。

3.2 版更改： ContextDecorator 的使用。

@contextlib.asynccontextmanager

类似于 contextmanager()，但创建一个 异步上下文管理器 。

该函数是一个装饰器，可用于为async with语句异步上下文管理器定义工厂函数，无需创建类或将__aenter__()和[ X209X] 方法。 它必须应用于 异步发电机 功能。

一个简单的例子：

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 版中的新功能。

contextlib.closing(thing)

返回一个上下文管理器，在块完成时关闭 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('https://www.python.org')) as page:
    for line in page:
        print(line)

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

contextlib.nullcontext(enter_result=None)

返回一个从 __enter__ 返回 enter_result 的上下文管理器，否则什么都不做。 它旨在用作可选上下文管理器的替代品，例如：

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 版中的新功能。

contextlib.suppress(*exceptions)

返回一个上下文管理器，如果它们发生在 with 语句的主体中，则抑制任何指定的异常，然后使用 with 语句结束后的第一个语句恢复执行。

与完全抑制异常的任何其他机制一样，此上下文管理器应仅用于覆盖非常特定的错误，其中已知静默继续执行程序是正确的做法。

例如：

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

这个上下文管理器是 可重入的 。

3.4 版中的新功能。

contextlib.redirect_stdout(new_target)

上下文管理器，用于临时将 sys.stdout 重定向到另一个文件或类似文件的对象。

此工具为输出硬连线到标准输出的现有函数或类增加了灵活性。

例如，help() 的输出通常被发送到 sys.stdout。 您可以通过将输出重定向到 io.StringIO 对象来捕获字符串中的输出。 替换流从 __enter__ 方法返回，因此可用作 with 语句的目标：

with redirect_stdout(io.StringIO()) as 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 的全局副作用意味着此上下文管理器不适合在库代码和大多数线程应用程序中使用。 它也对子流程的输出没有影响。 但是，对于许多实用程序脚本来说，它仍然是一种有用的方法。

这个上下文管理器是 可重入的 。

3.4 版中的新功能。

contextlib.redirect_stderr(new_target)

类似于 redirect_stdout() 但将 sys.stderr 重定向到另一个文件或类似文件的对象。

这个上下文管理器是 可重入的 。

3.5 版中的新功能。

class contextlib.ContextDecorator

使上下文管理器也可以用作装饰器的基类。

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

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

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 适用于整个函数，而不仅仅是它的一部分（并且保存缩进级别也很好）。

已经有基类的现有上下文管理器可以通过使用 ContextDecorator 作为混合类来扩展：

from contextlib import ContextDecorator

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

    def __exit__(self, *exc):
        return False

笔记

由于修饰函数必须能够被多次调用，底层上下文管理器必须支持在多个 with 语句中使用。 如果不是这种情况，则应使用在函数内具有显式 with 语句的原始构造。

3.2 版中的新功能。

class contextlib.ExitStack

一个上下文管理器，旨在使以编程方式组合其他上下文管理器和清理功能变得容易，尤其是那些可选的或由输入数据驱动的。

例如，可以在单个 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 语句的末尾显式或隐式地）调用这些回调的堆栈。 请注意，当上下文堆栈实例被垃圾收集时，回调不是 而不是 隐式调用。

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

由于注册的回调是以注册的相反顺序调用的，这最终表现为多个嵌套的 with 语句已与注册的回调集一起使用。 这甚至扩展到异常处理 - 如果内部回调抑制或替换异常，则外部回调将根据更新的状态传递参数。

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

3.3 版中的新功能。

enter_context(cm)

进入一个新的上下文管理器并将其 __exit__() 方法添加到回调堆栈中。 返回值是上下文管理器自己的 __enter__() 方法的结果。

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

push(exit)

将上下文管理器的 __exit__() 方法添加到回调堆栈。

由于 __enter__ 不是 而不是 调用，因此该方法可用于使用上下文管理器自己的 __exit__() 方法覆盖 __enter__() 实现的一部分。

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

通过返回真值，这些回调可以像上下文管理器 __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()

立即展开回调堆栈，以与注册相反的顺序调用回调。 对于注册的任何上下文管理器和退出回调，传入的参数将表明没有发生异常。

class contextlib.AsyncExitStack

一个 异步上下文管理器 ，类似于 ExitStack，它支持结合同步和异步上下文管理器，以及具有用于清理逻辑的协程。

close() 方法未实现，必须使用 aclose() 代替。

push_async_exit(exit)

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

push_async_callback(callback, /, *args, **kwds)

类似于 callback() 但需要一个协程函数。

继续 asynccontextmanager() 的示例：

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 版中的新功能。

例子和食谱

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

支持可变数量的上下文管理器

ExitStack 的主要用例是类文档中给出的用例：在单个 with 语句中支持可变数量的上下文管理器和其他清理操作。 可变性可能来自由用户输入驱动的上下文管理器的数量（例如打开用户指定的文件集合），或者来自一些可选的上下文管理器：

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 还可以很容易地使用 和 语句来管理本机不支持上下文管理协议的任意资源。

从 __enter__ 方法捕获异常

偶尔需要从 __enter__ 方法实现中捕获异常，without 无意中从 with 语句主体或上下文管理器的 __exit__ 方法捕获异常. 通过使用 ExitStack，上下文管理协议中的步骤可以稍微分开，以允许这样做：

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

实际上需要这样做很可能表明底层 API 应该提供一个直接的资源管理接口用于 try/except/finally 语句，但是并非所有 API 都在这方面设计得很好。 当上下文管理器是唯一提供的资源管理 API 时，ExitStack 可以更轻松地处理无法在 with 语句中直接处理的各种情况。

在 __enter__ 实现中进行清理

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

下面是一个为上下文管理器执行此操作的示例，该上下文管理器接受资源获取和释放函数以及可选的验证函数，并将它们映射到上下文管理协议：

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 和标志变量的任何使用

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

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

与任何基于 try 语句的代码一样，这可能会给开发和审查带来问题，因为设置代码和清理代码最终可能会被任意长的代码段分开。

ExitStack 可以在 with 语句的末尾注册一个回调来执行，然后决定跳过执行该回调：

from contextlib import ExitStack

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

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

如果一个特定的应用程序经常使用这个模式，它可以通过一个小的帮助类进一步简化：

from contextlib import ExitStack

class Callback(ExitStack):
    def __init__(self, callback, /, *args, **kwds):
        super().__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()

如果资源清理还没有整齐地捆绑到一个独立的函数中，那么仍然可以使用 ExitStack.callback() 的装饰器形式提前声明资源清理：

from contextlib import ExitStack

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

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

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

ContextDecorator 使得可以在普通的 with 语句和函数装饰器中使用上下文管理器。

例如，有时用可以跟踪进入时间和退出时间的记录器包装函数或语句组很有用。 与其为任务编写函数装饰器和上下文管理器，不如从 ContextDecorator 继承，在单个定义中提供这两种功能：

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)

此类的实例既可以用作上下文管理器：

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

也可以作为函数装饰器：

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

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

也可以看看

PEP 343 - “with”语句

Python with 语句的规范、背景和示例。

一次性使用、可重用和可重入的上下文管理器

大多数上下文管理器的编写方式意味着它们只能在 with 语句中有效使用一次。 这些一次性使用的上下文管理器必须在每次使用时重新创建 - 尝试第二次使用它们将触发异常或以其他方式无法正常工作。

这种常见的限制意味着通常建议直接在使用它们的 with 语句的标头中创建上下文管理器（如上面的所有使用示例所示）。

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

使用 contextmanager() 创建的上下文管理器也是一次性使用的上下文管理器，如果第二次尝试使用它们，将会抱怨底层生成器无法让步：

>>> 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

可重入上下文管理器

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

threading.RLock 是可重入上下文管理器的示例，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() 绝对不是线程安全的，因为它通过将 sys.stdout 绑定到不同的流来对系统状态进行全局修改。

可重用的上下文管理器

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

threading.Lock 是一个可重用但不可重入的上下文管理器的示例（对于可重入锁，必须使用 threading.RLock 代替）。

另一个可重用但不可重入的上下文管理器的例子是 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 语句中重用单个堆栈对象可以正常工作，但尝试嵌套它们将导致堆栈在最里面的 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