contextlib --- 為 with語句上下文提供的工具?

工具?

提供的函數(shù)和類：

class contextlib.AbstractContextManager?

一個(gè)為實(shí)現(xiàn)了 object.__enter__() 與 object.__exit__() 的類提供的 abstract base class。為 object.__enter__() 提供的一個(gè)默認(rèn)實(shí)現(xiàn)是返回 self 而 object.__exit__() 是一個(gè)默認(rèn)返回 None 的抽象方法。 參見 上下文管理器類型 的定義。

3.6 新版功能.

class contextlib.AbstractAsyncContextManager?

一個(gè)為實(shí)現(xiàn)了 object.__aenter__() 與 object.__aexit__() 的類提供的 abstract base class。 為 object.__aenter__() 提供的一個(gè)默認(rèn)實(shí)現(xiàn)是返回 self 而 object.__aexit__() 是一個(gè)默認(rèn)返回 None 的抽象方法。 參見 異步上下文管理器 的定義。

3.7 新版功能.

@contextlib.contextmanager?

這個(gè)函數(shù)是一個(gè) decorator ，它可以定義一個(gè)支持 with 語句上下文管理器的工廠函數(shù)， 而不需要?jiǎng)?chuàng)建一個(gè)類或區(qū) __enter__() 與 __exit__() 方法。

盡管許多對(duì)象原生支持使用 with 語句，但有些需要被管理的資源并不是上下文管理器，并且沒有實(shí)現(xiàn) close() 方法而不能使用 contextlib.closing 。

下面是一個(gè)抽象的示例，展示如何確保正確的資源管理:

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

被裝飾的函數(shù)在被調(diào)用時(shí)，必須返回一個(gè) generator 迭代器。 這個(gè)迭代器必須只 yield 一個(gè)值出來，這個(gè)值會(huì)被用在 with 語句中，綁定到 as 后面的變量，如果給定了的話。

當(dāng)生成器發(fā)生 yield 時(shí)，嵌套在 with 語句中的語句體會(huì)被執(zhí)行。 語句體執(zhí)行完畢離開之后，該生成器將被恢復(fù)執(zhí)行。 如果在該語句體中發(fā)生了未處理的異常，則該異常會(huì)在生成器發(fā)生 yield 時(shí)重新被引發(fā)。 因此，你可以使用 try...except...finally 語句來捕獲該異常（如果有的話），或確保進(jìn)行了一些清理。 如果僅出于記錄日志或執(zhí)行某些操作（而非完全抑制異常）的目的捕獲了異常，生成器必須重新引發(fā)該異常。 否則生成器的上下文管理器將向 with 語句指示該異常已經(jīng)被處理，程序?qū)⒘⒓丛?with 語句之后恢復(fù)并繼續(xù)執(zhí)行。

contextmanager() 使用 ContextDecorator 因此它創(chuàng)建的上下文管理器不僅可以用在 with 語句中，還可以用作一個(gè)裝飾器。當(dāng)它用作一個(gè)裝飾器時(shí)，每一次函數(shù)調(diào)用時(shí)都會(huì)隱式創(chuàng)建一個(gè)新的生成器實(shí)例（這使得 contextmanager() 創(chuàng)建的上下文管理器滿足了支持多次調(diào)用以用作裝飾器的需求，而非“一次性”的上下文管理器）。

在 3.2 版更改: ContextDecorator 的使用。

@contextlib.asynccontextmanager?

與 contextmanager() 類似，但創(chuàng)建的是 asynchronous context manager 。

這個(gè)函數(shù)是一個(gè) decorator ，它可以定義一個(gè)支持 async with 語句的異步上下文管理器的工廠函數(shù)， 而不需要?jiǎng)?chuàng)建一個(gè)類或區(qū)分 __aenter__() 與 __aexit__() 方法。它必須被作用在一個(gè) asynchronous generator 函數(shù)上

一個(gè)簡(jiǎn)單的示例：

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 新版功能.

Context managers defined with asynccontextmanager() can be used either as decorators or with async with statements:

import time
from contextlib import asynccontextmanager

@asynccontextmanager
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 ...

When used as a decorator, a new generator instance is implicitly created on each function call. This allows the otherwise "one-shot" context managers created by asynccontextmanager() to meet the requirement that context managers support multiple invocations in order to be used as decorators.

在 3.10 版更改: Async context managers created with asynccontextmanager() can be used as decorators.

contextlib.closing(thing)?

返回一個(gè)在語句塊執(zhí)行完成時(shí)關(guān)閉 things 的上下文管理器。這基本上等價(jià)于：

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)

而無需顯式地關(guān)閉 page 。 即使發(fā)生錯(cuò)誤，在退出 with 語句塊時(shí)， page.close() 也同樣會(huì)被調(diào)用。

contextlib.aclosing(thing)?

Return an async context manager that calls the aclose() method of thing upon completion of the block. This is basically equivalent to:

from contextlib import asynccontextmanager

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

Significantly, aclosing() supports deterministic cleanup of async generators when they happen to exit early by break or an exception. For example:

from contextlib import aclosing

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

This pattern ensures that the generator's async exit code is executed in the same context as its iterations (so that exceptions and context variables work as expected, and the exit code isn't run after the lifetime of some task it depends on).

3.10 新版功能.

contextlib.nullcontext(enter_result=None)?

返回一個(gè)從 __enter__ 返回 enter_result 的上下文管理器，除此之外不執(zhí)行任何操作。它旨在用于可選上下文管理器的一種替代，例如：

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

一個(gè)使用 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

It can also be used as a stand-in for 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 support was added.

contextlib.suppress(*exceptions)?

Return a context manager that suppresses any of the specified exceptions if they occur in the body of a with statement and then resumes execution with the first statement following the end of the with statement.

與完全抑制異常的任何其他機(jī)制一樣，該上下文管理器應(yīng)當(dāng)只用來抑制非常具體的錯(cuò)誤，并確保該場(chǎng)景下靜默地繼續(xù)執(zhí)行程序是通用的正確做法。

例如：

from contextlib import suppress

with suppress(FileNotFoundError):
    os.remove('somefile.tmp')

with suppress(FileNotFoundError):
    os.remove('someotherfile.tmp')

這段代碼等價(jià)于：

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 臨時(shí)重定向到一個(gè)文件或類文件對(duì)象的上下文管理器。

該工具給已有的將輸出硬編碼寫到 stdout 的函數(shù)或類提供了額外的靈活性。

For example, the output of help() normally is sent to sys.stdout. You can capture that output in a string by redirecting the output to an io.StringIO object. The replacement stream is returned from the __enter__ method and so is available as the target of the with statement:

with redirect_stdout(io.StringIO()) as f:
    help(pow)
s = f.getvalue()

如果要把 help() 的輸出寫到磁盤上的一個(gè)文件，重定向該輸出到一個(gè)常規(guī)文件：

with open('help.txt', 'w') as f:
    with redirect_stdout(f):
        help(pow)

如果要把 help() 的輸出寫到 sys.stderr ：

with redirect_stdout(sys.stderr):
    help(pow)

需要注意的點(diǎn)在于， sys.stdout 的全局副作用意味著此上下文管理器不適合在庫(kù)代碼和大多數(shù)多線程應(yīng)用程序中使用。它對(duì)子進(jìn)程的輸出沒有影響。不過對(duì)于許多工具腳本而言，它仍然是一個(gè)有用的方法。

該上下文管理器是 reentrant 。

3.4 新版功能.

contextlib.redirect_stderr(new_target)?

與 redirect_stdout() 類似，不過是將 sys.stderr 重定向到一個(gè)文件或類文件對(duì)象。

該上下文管理器是 reentrant 。

3.5 新版功能.

contextlib.chdir(path)?

Non parallel-safe context manager to change the current working directory. As this changes a global state, the working directory, it is not suitable for use in most threaded or async contexts. It is also not suitable for most non-linear code execution, like generators, where the program execution is temporarily relinquished -- unless explicitely desired, you should not yield when this context manager is active.

This is a simple wrapper around chdir(), it changes the current working directory upon entering and restores the old one on exit.

該上下文管理器是 reentrant 。

3.11 新版功能.

class contextlib.ContextDecorator?

一個(gè)使上下文管理器能用作裝飾器的基類。

與往常一樣，繼承自 ContextDecorator  的上下文管理器必須實(shí)現(xiàn) __enter__ 與 __exit__ 。即使用作裝飾器， __exit__ 依舊會(huì)保持可能的異常處理。

ContextDecorator 被用在 contextmanager() 中，因此你自然獲得了這項(xiàng)功能。

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

這個(gè)改動(dòng)只是針對(duì)如下形式的一個(gè)語法糖：

def f():
    with cm():
        # Do stuff

ContextDecorator 使得你可以這樣改寫：

@cm()
def f():
    # Do stuff

這能清楚地表明， cm 作用于整個(gè)函數(shù)，而不僅僅是函數(shù)的一部分（同時(shí)也能保持不錯(cuò)的縮進(jìn)層級(jí)）。

現(xiàn)有的上下文管理器即使已經(jīng)有基類，也可以使用 ContextDecorator 作為混合類進(jìn)行擴(kuò)展：

from contextlib import ContextDecorator

class mycontext(ContextBaseClass, ContextDecorator):
    def __enter__(self):
        return self

    def __exit__(self, *exc):
        return False

備注

由于被裝飾的函數(shù)必須能夠被多次調(diào)用，因此對(duì)應(yīng)的上下文管理器必須支持在多個(gè) with 語句中使用。如果不是這樣，則應(yīng)當(dāng)使用原來的具有顯式 with 語句的形式使用該上下文管理器。

3.2 新版功能.

class contextlib.AsyncContextDecorator?

Similar to ContextDecorator but only for asynchronous functions.

Example of 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?

該上下文管理器的設(shè)計(jì)目標(biāo)是使得在編碼中組合其他上下文管理器和清理函數(shù)更加容易，尤其是那些可選的或由輸入數(shù)據(jù)驅(qū)動(dòng)的上下文管理器。

例如，通過一個(gè)如下的 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

The __enter__() method returns the ExitStack instance, and performs no additional operations.

每個(gè)實(shí)例維護(hù)一個(gè)注冊(cè)了一組回調(diào)的棧，這些回調(diào)在實(shí)例關(guān)閉時(shí)以相反的順序被調(diào)用（顯式或隱式地在 with 語句的末尾）。請(qǐng)注意，當(dāng)一個(gè)棧實(shí)例被垃圾回收時(shí)，這些回調(diào)將 不會(huì) 被隱式調(diào)用。

通過使用這個(gè)基于棧的模型，那些通過 __init__ 方法獲取資源的上下文管理器（如文件對(duì)象）能夠被正確處理。

由于注冊(cè)的回調(diào)函數(shù)是按照與注冊(cè)相反的順序調(diào)用的，因此最終的行為就像多個(gè)嵌套的 with 語句用在這些注冊(cè)的回調(diào)函數(shù)上。這個(gè)行為甚至擴(kuò)展到了異常處理：如果內(nèi)部的回調(diào)函數(shù)抑制或替換了異常，則外部回調(diào)收到的參數(shù)是基于該更新后的狀態(tài)得到的。

這是一個(gè)相對(duì)底層的 API，它負(fù)責(zé)正確處理?xiàng)＠锘卣{(diào)退出時(shí)依次展開的細(xì)節(jié)。它為相對(duì)高層的上下文管理器提供了一個(gè)合適的基礎(chǔ)，使得它能根據(jù)應(yīng)用程序的需求使用特定方式操作棧。

3.3 新版功能.

enter_context(cm)?

Enters a new context manager and adds its __exit__() method to the callback stack. The return value is the result of the context manager's own __enter__() method.

These context managers may suppress exceptions just as they normally would if used directly as part of a with statement.

在 3.11 版更改: Raises TypeError instead of AttributeError if cm is not a context manager.

push(exit)?

Adds a context manager's __exit__() method to the callback stack.

As __enter__ is not invoked, this method can be used to cover part of an __enter__() implementation with a context manager's own __exit__() method.

If passed an object that is not a context manager, this method assumes it is a callback with the same signature as a context manager's __exit__() method and adds it directly to the callback stack.

By returning true values, these callbacks can suppress exceptions the same way context manager __exit__() methods can.

The passed in object is returned from the function, allowing this method to be used as a function decorator.

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

Accepts an arbitrary callback function and arguments and adds it to the callback stack.

Unlike the other methods, callbacks added this way cannot suppress exceptions (as they are never passed the exception details).

The passed in callback is returned from the function, allowing this method to be used as a function decorator.

pop_all()?

Transfers the callback stack to a fresh ExitStack instance and returns it. No callbacks are invoked by this operation - instead, they will now be invoked when the new stack is closed (either explicitly or implicitly at the end of a with statement).

For example, a group of files can be opened as an "all or nothing" operation as follows:

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()?

Immediately unwinds the callback stack, invoking callbacks in the reverse order of registration. For any context managers and exit callbacks registered, the arguments passed in will indicate that no exception occurred.

class contextlib.AsyncExitStack?

An asynchronous context manager, similar to ExitStack, that supports combining both synchronous and asynchronous context managers, as well as having coroutines for cleanup logic.

The close() method is not implemented, aclose() must be used instead.

coroutine enter_async_context(cm)?

Similar to enter_context() but expects an asynchronous context manager.

在 3.11 版更改: Raises TypeError instead of AttributeError if cm is not an asynchronous context manager.

push_async_exit(exit)?

Similar to push() but expects either an asynchronous context manager or a coroutine function.

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

Similar to callback() but expects a coroutine function.

coroutine aclose()?

Similar to close() but properly handles awaitables.

Continuing the example for 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 新版功能.

例子和配方?

This section describes some examples and recipes for making effective use of the tools provided by contextlib.

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

stack = ExitStack()
try:
    x = stack.enter_context(cm)
except Exception:
    # handle __enter__ exception
else:
    with stack:
        # Handle normal case

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

cleanup_needed = True
try:
    result = perform_operation()
    if result:
        cleanup_needed = False
finally:
    if cleanup_needed:
        cleanup_resources()

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

from contextlib import ExitStack

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

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

Single use, reusable and reentrant context managers?

Most context managers are written in a way that means they can only be used effectively in a with statement once. These single use context managers must be created afresh each time they're used - attempting to use them a second time will trigger an exception or otherwise not work correctly.

This common limitation means that it is generally advisable to create context managers directly in the header of the with statement where they are used (as shown in all of the usage examples above).

Files are an example of effectively single use context managers, since the first with statement will close the file, preventing any further IO operations using that file object.

Context managers created using contextmanager() are also single use context managers, and will complain about the underlying generator failing to yield if an attempt is made to use them a second time:

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

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

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

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