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 版中的新功能。
使用 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)
返回一个上下文管理器,在块完成时关闭 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()
。
- class contextlib.aclosing(thing)
返回一个异步上下文管理器,该管理器在块完成时调用 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__
返回 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
它也可以用作 异步上下文管理器 的替代品:
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 版更改:添加了 异步上下文管理器 支持。
- 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.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 语句的末尾显式或隐式地)调用这些回调的堆栈。 请注意,当上下文堆栈实例被垃圾收集时,回调不是 而不是 隐式调用。
使用此堆栈模型,以便可以正确处理在其
__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
语句。
一次性使用、可重用和可重入的上下文管理器
大多数上下文管理器的编写方式意味着它们只能在 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