contextlib — Утилиты для контекстов оператора 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

Эта функция является декоратором, который может использоваться для определения функции-фабрики для with контекстных менеджеров, без необходимости создания класса или отдельных методов __enter__() и __exit__().

Хотя многие объекты изначально поддерживают использование в операторах with, иногда необходимо управлять ресурсом, который сам по себе не является контекстным менеджером и не реализует метод close() для использования с contextlib.closing

Абстрактный пример, который обеспечит правильное управление ресурсами:

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

Функция, к которой применяется декоратор, должна возвращать генератор-итератор при вызове. Этот итератор должен сгенерировать ровно одно значение, которое будет привязано к целевым переменным в with оператора with, если таковые имеются.

В момент генерации значения блок, вложенный в оператор with, выполняется. Генератор возобновляется после выхода из блока. Если возникает необработанное исключение в блоке, оно переизлучается внутри генератора в точке, где произошло yield. Таким образом, вы можете использовать оператор try…except…finally, чтобы перехватить ошибку (если она есть) или гарантировать выполнение некоторой очистки. Если исключение перехватывается только для того, чтобы его записать в журнал или выполнить какое-либо действие (а не для того, чтобы его полностью подавить), генератор должен повторно излучать это исключение. В противном случае контекстный менеджер генератора укажет оператору with , что исключение обработано, и выполнение продолжится с оператора, следующего непосредственно за оператором with

contextmanager() использует ContextDecorator, поэтому создаваемые им контекстные менеджеры могут использоваться в качестве декораторов, а также в операторах with. При использовании в качестве декоратора новый экземпляр генератора неявно создается при каждом вызове функции (это позволяет контекстным менеджерам, созданным contextmanager(), которые в противном случае «одноразовые», удовлетворять требованию, что контекстные менеджеры поддерживают несколько вызовов, чтобы их можно было использовать в качестве декораторов).

Изменено в версии 3.2: Использование ContextDecorator.

@contextlib.asynccontextmanager

Аналогично contextmanager(), но создаёт асинхронный контекстный менеджер.

Эта функция является декоратором, который можно использовать для определения фабричной функции для асинхронных контекстных менеджеров оператора async with без необходимости создания класса или отдельных методов __aenter__() и __aexit__(). Она должна применяться к функции асинхронного генератора.

Простой пример:

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.

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('http://www.python.org')) as page:
    for line in page:
        print(line)

без необходимости явно закрывать page. Даже если произойдёт ошибка, page.close() будет вызван при выходе из блока with.

contextlib.nullcontext(enter_result=None)

Возвращает контекстный менеджер, который возвращает enter_result из __enter__, но при этом ничего не делает. Он предназначен для использования в качестве замены для необязательного контекстного менеджера, например:

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 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 в другой файл или объект, подобный файлу.

Этот инструмент добавляет гибкость существующим функциям или классам, вывод которых жёстко привязан к 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 означает, что этот контекстный менеджер не подходит для использования в библиотечном коде и большинстве потоковых приложений. Он также не влияет на вывод подпроцессов. Однако это всё ещё полезный подход для многих утилитарных скриптов.

Этот контекстный менеджер реентерабелен.

Новая версия 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.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__ не вызывается, этот метод может использоваться для покрытия части реализации __enter__() собственным методом менеджера контекста __exit__().

Если передан объект, который не является менеджером контекста, этот метод предполагает, что это обратный вызов с той же сигнатурой, что и метод __exit__() менеджера контекста, и добавляет его напрямую в стек обратных вызовов.

Возвращая значения true, эти обратные вызовы могут подавлять исключения так же, как могут методы менеджеров контекста __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() вместо него.

enter_async_context(cm)

Аналогично enter_context(), но ожидает асинхронный менеджер контекста.

push_async_exit(exit)

Аналогично push(), но ожидает либо асинхронный менеджер контекста, либо функцию сопрограммы.

push_async_callback(callback, *args, **kwds)

Аналогично callback(), но ожидает функцию сопрограммы.

aclose()

Аналогично close(), но корректно обрабатывает awaitables.

Продолжение примера для 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.

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

Одноразовые, многоразовые и рекурсивные контекстные менеджеры

Большинство контекстных менеджеров написаны так, что они могут быть эффективно использованы в операторе with только один раз. Эти одноразовые контекстные менеджеры должны создаваться заново каждый раз при их использовании — попытка использовать их во второй раз вызовет исключение или иначе не будет работать правильно.

Это общее ограничение означает, что, как правило, рекомендуется создавать контекстные менеджеры непосредственно в заголовке оператора with, где они используются (как показано во всех примерах использования выше).

Файлы — это пример эффективных одноразовых контекстных менеджеров, так как первый оператор with закроет файл, предотвращая дальнейшие операции ввода-вывода с использованием этого объекта файла.

Контекстные менеджеры, созданные с использованием 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

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