contextlib ---公用设施 with -语句上下文

源代码： Lib/contextlib.py

---

此模块为涉及 with 语句。有关更多信息，请参阅 上下文管理器类型 和 使用语句上下文管理器 .

公用事业

提供的功能和类：

class contextlib.AbstractContextManager

安 abstract base class 对于实现 object.__enter__() 和 object.__exit__() . 的默认实现 object.__enter__() 提供返回的 self 虽然 object.__exit__() 是默认返回的抽象方法 None . 另见定义 上下文管理器类型 .

3.6 新版功能.

class contextlib.AbstractAsyncContextManager

安 abstract base class 对于实现 object.__aenter__() 和 object.__aexit__() . 的默认实现 object.__aenter__() 提供返回的 self 虽然 object.__aexit__() 是默认返回的抽象方法 None . 另见定义 异步上下文管理器 .

3.7 新版功能.

@contextlib.contextmanager

此函数是 decorator 可用于定义factory函数 with 语句上下文管理器，无需创建类或单独 __enter__() 和 __exit__() 方法。

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

为确保正确的资源管理，抽象示例如下：

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 -调用迭代器时。此迭代器必须只生成一个值，该值将绑定到 with 语句的 as 子句，如果有的话。

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

contextmanager() 使用 ContextDecorator 因此，它创建的上下文管理器既可以用作修饰器，也可以用于 with 声明。当用作装饰器时，在每个函数调用上隐式创建一个新的生成器实例（这允许由 contextmanager() 以满足上下文管理器支持多个调用以用作装饰器的要求）。

在 3.2 版更改: 使用 ContextDecorator .

@contextlib.asynccontextmanager

类似 contextmanager() ，但创建 asynchronous context manager .

此函数是 decorator 可用于定义factory函数 async with 语句异步上下文管理器，无需创建类或单独 __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 新版功能.

上下文管理器定义为 asynccontextmanager() 既可以作为装饰品使用，也可以与 async with 声明：

import time

async def timeit():
    now = time.monotonic()
    try:
        yield
    finally:
        print(f'it took {time.monotonic() - now}s to run')

 @timeit()
 async def main():
     # ... async code ...

当用作装饰器时，每个函数调用都会隐式创建一个新的生成器实例。这允许在其他情况下由创建的“一次性”上下文管理器 asynccontextmanager() 以满足上下文管理器支持多个调用以便用作装饰器的要求。

在 3.10 版更改: 使用创建的异步上下文管理器 asynccontextmanager() 可以作为装饰品使用。

contextlib.closing(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 . 即使发生错误， page.close() 当 with 块已退出。

class contextlib.aclosing(thing)

返回一个异步上下文管理器，该管理器调用 aclose() 一种方法 一件事 在挡路建成后。这基本上相当于：

from contextlib import asynccontextmanager

@asynccontextmanager
async def aclosing(thing):
    try:
        yield thing
    finally:
        await thing.aclose()

值得注意的是， aclosing() 当异步生成器碰巧通过以下方式提前退出时，支持确定性清除 break 或者是例外。例如：：

from contextlib import aclosing

async with aclosing(my_generator()) as values:
    async for value in values:
        if value == 42:
            break

此模式确保生成器的异步退出代码在与其迭代相同的上下文中执行(以便异常和上下文变量按预期工作，并且退出代码不会在它所依赖的某些任务的生存期之后运行)。

3.10 新版功能.

contextlib.nullcontext(enter_result=None)

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

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

它还可以用作 asynchronous context managers ：：

async def send_http(session=None):
   if not session:
       # If no http session, create it with aiohttp
       cm = aiohttp.ClientSession()
   else:
       # Caller is responsible for closing the session
       cm = nullcontext(session)

   async with cm as session:
       # Send http requests with session

3.7 新版功能.

在 3.10 版更改: asynchronous context manager 添加了支持。

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

此上下文管理器是 reentrant .

3.4 新版功能.

contextlib.redirect_stdout(new_target)

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

这个工具增加了现有函数或类的灵活性，这些函数或类的输出是硬连接到stdout的。

例如，输出 help() 通常发送到 sys.stdout . 通过将输出重定向到 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 这意味着上下文管理器不适合在库代码和大多数线程应用程序中使用。它对子流程的输出也没有影响。但是，对于许多实用程序脚本来说，它仍然是一种有用的方法。

此上下文管理器是 reentrant .

3.4 新版功能.

contextlib.redirect_stderr(new_target)

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

此上下文管理器是 reentrant .

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

类似于 ContextDecorator 但仅适用于异步函数。

示例： AsyncContextDecorator ：：

from asyncio import run
from contextlib import AsyncContextDecorator

class mycontext(AsyncContextDecorator):
    async def __aenter__(self):
        print('Starting')
        return self

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

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

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

3.10 新版功能.

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 声明）。请注意，回调是 not 当上下文堆栈实例被垃圾收集时隐式调用。

使用此堆栈模型可以使上下文管理器在 __init__ 方法（如文件对象）可以正确处理。

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

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

3.3 新版功能.

enter_context(cm)

输入新的上下文管理器并添加其 __exit__() 方法调用堆栈。返回值是上下文管理器自己的结果 __enter__() 方法。

如果直接用作 with 语句。

push(exit)

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

AS __enter__ 是 not 调用时，此方法可用于覆盖 __enter__() 使用上下文管理器自己的实现 __exit__() 方法。

如果传递的对象不是上下文管理器，则此方法假定它是具有与上下文管理器相同签名的回调 __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

安 asynchronous context manager ，类似于 ExitStack 它支持同步和异步上下文管理器的组合，以及具有用于清除逻辑的协程。

这个 close() 方法未实现， aclose() 必须使用。

enter_async_context(cm)

类似 enter_context() 但需要一个异步上下文管理器。

push_async_exit(exit)

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

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

类似 callback() 但是需要协同作用。

aclose()

类似 close() 但要妥善处理等待的事情。

继续示例 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 也使它很容易使用 with 用于管理本机不支持上下文管理协议的任意资源的语句。

从中捕获异常 __enter__ 方法

有时需要从 __enter__ 方法实施， 没有 无意中从 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(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()

如果资源清理尚未整齐地捆绑到一个独立的函数中，那么仍然可以使用 ExitStack.callback() 要提前声明资源清理，请执行以下操作：

from contextlib import ExitStack

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

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

将上下文管理器用作函数修饰器

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

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

还要注意，可重入性是 not 和线程安全一样。 redirect_stdout() 例如，绝对不是线程安全的，因为它通过绑定对系统状态进行全局修改。 sys.stdout 去另一条小溪。

可重用上下文管理器

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

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

另一个可重用但不可重入的上下文管理器示例是 ExitStack ，正如它所调用的 all 在保留WITH语句时当前注册的回调，无论这些回调添加到何处：：

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