On this page
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 器类型的定义。
- 实现抽象 Base Class和object.exit()的类的抽象 Base Class。提供了object.enter()的默认实现,该实现返回
3.6 版的新Function。
- 类别
contextlib.
AbstractAsyncContextManager
- 实现抽象 Base Class和object.aexit()的类的抽象 Base Class。提供了object.aenter()的默认实现,该实现返回
self
,而object.aexit()是抽象方法,默认情况下返回None
。另请参见异步上下文 Management 器的定义。
- 实现抽象 Base Class和object.aexit()的类的抽象 Base Class。提供了object.aenter()的默认实现,该实现返回
3.7 版中的新Function。
@
contextlib.
contextmanager
尽管许多对象本机支持在 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 器的替代,例如:
- 返回一个上下文 Management 器,该上下文 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
Note
由于修饰的函数必须能够被多次调用,因此基础上下文 Management 器必须支持在多个with语句中使用。如果不是这种情况,则应使用函数内部带有显式with
语句的原始构造。
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 *)
这些上下文 Management 器可以像直接将它们用作with语句的一部分一样抑制异常。
push
(退出)- 将上下文 Management 器的exit()方法添加到回调堆栈。
由于未调用__enter__
,因此此方法可用于使用上下文 Management 器自己的exit()方法覆盖enter()实现的一部分。
如果传递的对象不是上下文 Management 器,则此方法假定它是一个具有与上下文 Management 器的exit()方法相同签名的回调,并将其直接添加到回调堆栈中。
pass返回真值,这些回调可以像上下文 Management 器exit()方法一样抑制异常。
传入的对象从函数返回,从而允许将此方法用作函数装饰器。
callback
((* callback *, *args , * kwds *)- 接受任意的回调函数和参数,并将其添加到回调堆栈中。
与其他方法不同,以这种方式添加的回调无法抑制异常(因为它们从未传递过异常详细信息)。
传入的回调从函数返回,从而允许将此方法用作函数装饰器。
例如,可以按“全部或全部”操作打开一组文件,如下所示:
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