You are here:  Home » Python » – 用于-Python运行时服务的实用程序(Python教程)(参考资料)

contextlibwith-statement contexts

源代码:LIB / contextlib.py


该模块提供了涉及with声明。有关更多信息,请参阅上下文管理器类型使用语句上下文管理器.

实用程序

提供的函数和类:

class contextlib.AbstractContextManager

abstract base class 用于实现object.__enter__()object.__exit__()。提供了object.__enter__()defaultimplementation,它返回selfobject.__exit__()是一个抽象的方法,默认情况下返回None。另见上下文管理器类型的定义.

版本3.6.

class contextlib.AbstractAsyncContextManager

中的新增抽象基类用于实现object.__aenter__()object.__aexit__()的类。为object.__aenter__()提供了一个defaultimplementation,它返回selfobject.__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语句中执行。然后在块退出后恢复生成器。如果块中发生未处理的异常,则在生成器发生的点处将其重新生成。因此,您可以使用tryexceptfinally用于传递错误(如果有)的语句,或确保进行一些清理。如果仅为了记录异常或执行某些操作(而不是完全对其进行压缩)而捕获异常,则生成器必须重新加载该异常。否则,生成器上下文管理器将向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处理,即使用作装饰器时也是如此.

ContextDecoratorcontextmanager()使用,所以你自动得到这个功能.

示例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__()方法一样.

传入的对象从函数返回,允许这个方法用作函数装饰器.

callbackcallback, *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

立即展开回调堆栈,以反向的注册顺序调用回调。对于任何已注册的上下文管理器exitcallbacks,传入的参数将表明没有发生异常.

class contextlib.AsyncExitStack

一个异步上下文管理器, 相近 ExitStack,它支持同步和同步上下文管理器的组合,以及具有清理逻辑的协同程序.

close()方法没有实现,aclose()必须使用它.

enter_async_context (cm

enter_context()类似,但需要一个异步的contextmanager.

push_async_exitexit

相近 push()但是要求异步上下文管理和协同功能.

push_async_callbackcallback, *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的主要用例是给出的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的规范,背景和示例withstatement.

单独使用,可重用和可重入的上下文管理器

大多数上下文管理器都是用这意味着它们只能在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