– 用于-Python运行时服务的实用程序(Python教程)(参考资料)
contextlib
– with
-statement contexts
源代码:LIB / contextlib.py
该模块提供了涉及with
声明。有关更多信息,请参阅上下文管理器类型和使用语句上下文管理器.
实用程序
提供的函数和类:
- class
contextlib.
AbstractContextManager
-
abstract base class 用于实现
object.__enter__()
和object.__exit__()
。提供了object.__enter__()
的defaultimplementation,它返回self
而object.__exit__()
是一个抽象的方法,默认情况下返回None
。另见上下文管理器类型的定义.版本3.6.
- class
contextlib.
AbstractAsyncContextManager
-
中的新增抽象基类用于实现
object.__aenter__()
和object.__aexit__()
的类。为object.__aenter__()
提供了一个defaultimplementation,它返回self
而object.__aexit__()
是一个抽象的方法,默认情况下返回None
。另见异步上下文管理器的定义.版本3.7.
@
contextlib.
contextmanager
-
的新功能这个功能是装饰可用于定义
with
语句上下文管理器的工厂函数,无需创建类或单独的__enter__()
和__exit__()
methods.本机支持在with语句中使用,有时需要管理源本身不是上下文管理器,并且没有实现
close()
方法用于contextlib.closing
一个抽象的例子以下是确保正确的资源管理:
from contextlib import contextmanager@contextmanagerdef 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
正在装饰的函数必须返回一个生成器 – 调用时的迭代器。这个迭代器必须只生成一个值,它将绑定到
with
语句的as
子句中的目标,如果有的话在生成器产生的点上,块嵌套在
with
语句中执行。然后在块退出后恢复生成器。如果块中发生未处理的异常,则在生成器发生的点处将其重新生成。因此,您可以使用try
…except
…finally
用于传递错误(如果有)的语句,或确保进行一些清理。如果仅为了记录异常或执行某些操作(而不是完全对其进行压缩)而捕获异常,则生成器必须重新加载该异常。否则,生成器上下文管理器将向with
语句指示已处理异常,并且执行将在with
语句之后立即恢复状态.contextmanager()
使用ContextDecorator
所以上下文管理器创建可以用作装饰器以及with
语句。当用作装饰器时,一个新的生成器实例被隐式地创建一个函数调用(这允许否则“一次性”上下文由contextmanager()
创建的管理器,以满足上下文管理器支持多个调用以便用作装饰器的要求。)在版本3.2中更改:使用
ContextDecorator
.
@
contextlib.
asynccontextmanager
-
类似于
contextmanager()
,但创建一个异步上下文管理器.这个函数是装饰器可用于为
async with
语句异步上下文管理器定义工厂函数,而无需创建类或单独的__aenter__()
和__aexit__()
方法。它必须应用于异步发电机功能一个简单的例子:
from contextlib import asynccontextmanager@asynccontextmanagerasync 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@contextmanagerdef closing(thing): try: yield thing finally: thing.close()
并且让你编写这样的代码:
from contextlib import closingfrom urllib.request import urlopenwith closing(urlopen("http://www.python.org")) as page: for line in page: print(line)
而不需要明确地关闭
page
。即使发生错误,当page.close()
块退出时也会调用with
.
contextlib.
nullcontext
(enter_result=None)-
返回上下文从enter_result返回
__enter__
的经理,但是没有做什么。它可以用作anoptional上下文管理器的替身,例如: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 suppresswith suppress(FileNotFoundError): os.remove("somefile.tmp")with suppress(FileNotFoundError): os.remove("someotherfile.tmp")
这段代码相当于:
try: os.remove("somefile.tmp")except FileNotFoundError: passtry: os.remove("someotherfile.tmp")except FileNotFoundError: pass
这个上下文管理器是可重入.
版本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
意味着thiscontext管理器不适合在库代码和大多数线程应用程序中使用。它对子进程的输出也没有影响。但是,它仍然是许多实用程序脚本的有用方法.这个上下文管理器可重入.
版本3.4.
contextlib.
redirect_stderr
(new_target)-
与
redirect_stdout()
类似但是将sys.stderr
重定向到另一个文件或类似文件的对象.这个上下文管理器是reentrant .
版本3.5中的新功能.
- class
contextlib.
ContextDecorator
-
一个基类,它使上下文管理器也可以用作装饰器.
继承自
ContextDecorator
的文本管理器必须实施__enter__
和__exit__
像平常一样。__exit__
保留了它的optionalexception处理,即使用作装饰器时也是如此.ContextDecorator
被contextmanager()
使用,所以你自动得到这个功能.示例
ContextDecorator
:from contextlib import ContextDecoratorclass 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()StartingThe bit in the middleFinishing>>> with mycontext():... print("The bit in the middle")...StartingThe bit in the middleFinishing
这个改变只是以下形式的任何构造的语法糖:
def f(): with cm(): # Do stuff
ContextDecorator
让你改为写:@cm()def f(): # Do stuff
它清楚地表明
cm
适用于整个函数,而不是只是它的一部分(并且保存缩进级别也很好).已经有基类的现有上下文管理器可以通过使用
ContextDecorator
作为mixin类来扩展:from contextlib import ContextDecoratorclass mycontext(ContextBaseClass, ContextDecorator): def __enter__(self): return self def __exit__(self, *exc): return False
注意
由于装饰函数必须能够被多次调用,因此下层上下文管理器必须支持多个
with
语句。如果不是这种情况,那么应该使用函数内部带有显式with
语句的原始构造.新版本3.2.
- class
contextlib.
ExitStack
-
设计的上下文管理器使其易于以编程方式组合其他上下文管理器和清理函数,尤其是可选或由输入数据驱动的文件.
例如,一组文件可以在一个句子中轻松处理,如下所示:
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__()
方法添加到回调堆栈中。返回值是contextmanager自己的__enter__()
方法的结果如果直接用作
with
声明的一部分,这些上下文管理器可以像正常情况一样压制异常.
push
(exit)-
添加一个上下文管理器
__exit__()
回调堆栈的方法对于
__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.
- class
contextlib.
AsyncExitStack
-
一个异步上下文管理器, 相近
ExitStack
,它支持同步和同步上下文管理器的组合,以及具有清理逻辑的协同程序.close()
方法没有实现,aclose()
必须使用它.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
的主要用例是给出的classdocumentation:在单个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__
方法实现中捕获异常,without无意中从with
语句体或上下文管理器的__exit__
方法中捕获异常。通过使用 ExitStack
上下文管理协议中的步骤可以稍微分开,以便允许:
stack = ExitStack()try: x = stack.enter_context(cm)except Exception: # handle __enter__ exceptionelse: with stack: # Handle normal case
实际上需要这样做可能表明底层API应该提供一个直接的资源管理界面,用于try
/except
/finally
声明,但并非所有API都在这方面设计得很好。当上下文管理器只提供资源管理API时,ExitStack
可以使用iteasier来处理with
声明。
清理__enter__
实施
如ExitStack.push()
如果__enter__()
实施中的latersteps失败,这种方法可以用于清理已经分配的资源.
以下是为接受资源获取和释放功能的上下文管理器以及可选的验证功能执行此操作的示例,并将它们映射到上下文管理协议:
from contextlib import contextmanager, AbstractContextManager, ExitStackclass 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
带有flag变量的声明,表示是否finally
应该执行条款。在它最简单的形式(只能通过在except
子句中使用而不能处理),它看起来像这样:
cleanup_needed = Truetry: result = perform_operation() if result: cleanup_needed = Falsefinally: if cleanup_needed: cleanup_resources()
和任何try
基于语句的代码,这可能会导致开发和审查的问题,因为设置代码和清理代码最终会被任意长的代码段分开.
ExitStack
可以在with
语句的末尾注册一个回调forexecution,然后决定skipexecuting该回调:
from contextlib import ExitStackwith ExitStack() as stack: stack.callback(cleanup_resources) result = perform_operation() if result: stack.pop_all()
这允许预先清除预期的清理行为,而不是需要单独的标志变量.
如果一个特定的应用程序经常使用这个模式,它可以通过一个小帮助程序类进一步简化:
from contextlib import ExitStackclass 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()
的装饰器形式来声明资源清理inadvance:
from contextlib import ExitStackwith ExitStack() as stack: @stack.callback def cleanup_resources(): ... result = perform_operation() if result: stack.pop_all()
由于装饰器的方式协议工作,一个回调函数声明这种方式不能采取任何参数。相反,要释放的任何资源都必须作为闭包变量来访问.
使用上下文管理器作为函数装饰器
ContextDecorator
使得可以在普通的with
中使用上下文管理器语句以及函数decorator.
例如,用记录器包装函数或语句组有时很有用,它可以跟踪输入的时间和退出的时间。继承自ContextDecorator
而不是为任务编写函数装饰器和上下文管理器,而是在单个定义中提供两种功能:
from contextlib import ContextDecoratorimport logginglogging.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
statement.
单独使用,可重用和可重入的上下文管理器
大多数上下文管理器都是用这意味着它们只能在with
语句中有效使用一次。这些单个usecontext管理器必须在每次使用时重新创建 – 尝试再次使用它们会触发异常或者其他方式无法正常工作.
这个常见的限制意味着通常建议直接创建上下文管理器在with
语句的标题中使用它们(如上面的所有用法示例所示).
文件是有效的单一用户上下文管理器的一个例子,因为第一个with
语句将关闭文件,阻止使用该文件对象进行任何进一步的IO操作.
使用contextmanager()
创建的文本管理器也是单用的文本管理器,如果第二次尝试使用它们,就会抱怨底层生成器无法生成:
>>> from contextlib import contextmanager>>> @contextmanager... def singleuse():... print("Before")... yield... print("After")...>>> cm = singleuse()>>> with cm:... pass...BeforeAfter>>> with cm:... pass...Traceback (most recent call last): ...RuntimeError: generator didn"t yield
可重入上下文管理器
更复杂的上下文管理器可能是“可重入的””。这些上下文管理器不仅可以用在多个with
语句中,还可以用inside 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 stdoutThis is also written to the stream
真实世界的重入例子更可能涉及多个函数相互调用,因此比这个例子要复杂得多.
还注意到可重入的是not与线程安全相同的事情.redirect_stdout()
例如,绝对不是线程安全的,因为它通过将sys.stdout
绑定到不同的流来对系统状态进行全局修改.
可重用的上下文管理器
与单一使用和重入上下文管理器不同的是“可重用”的上下文管理器(或者,完全明确的,“可重用但不可靠”的上下文管理器,因为可重入的上下文管理器是可以使用的)。这些上下文管理器支持多次使用,但如果已经在contains with statement中使用了特定的上下文管理器实例,则会失败(或者无法正常工作)
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 contextCallback: from first context>>> with stack:... stack.callback(print, "Callback: from second context")... print("Leaving second context")...Leaving second contextCallback: 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 contextCallback: from inner contextCallback: from outer contextLeaving outer context
正如示例中的输出所示,重复使用多个语句中的单个堆栈对象可以正常工作,但是尝试嵌套它们会导致堆栈在最里面的句子结束时被清除,这不太可能是理想的行为.
使用单独的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 contextCallback: from inner contextLeaving outer contextCallback: from outer context