contextlib – Утилиты для контекстов оператора with¶contextlib – Utilities for with-statement contexts

Утилиты¶Utilities

Предоставляемые функции и классы:

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):
    # Код для получения ресурса, например:
    resource = acquire_resource(*args, **kwds)
    try:
        yield resource
    finally:
        # Код для освобождения ресурса, например:
        release_resource(resource)

Затем функцию можно использовать следующим образом:

>>> with managed_resource(timeout=3600) as resource:
...     # Ресурс освобождается в конце этого блока,
...     # даже если код в блоке вызывает исключение

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

В момент, где генератор делает yield, выполняется блок, вложенный в оператор 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.

Контекстные менеджеры, определенные с помощью asynccontextmanager(), могут использоваться как в качестве декораторов, так и с операторами async with:

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():
    # ... асинхронный код ...

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

Изменено в версии 3.10: Асинхронные контекстные менеджеры, созданные с помощью asynccontextmanager(), могут использоваться в качестве декораторов.

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

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

Примечание

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

contextlib.aclosing(thing)¶

Возвращает асинхронный контекстный менеджер, который вызывает метод aclose() объекта thing после завершения блока. По сути эквивалентно следующему:

from contextlib import asynccontextmanager

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

Примечательно, что aclosing() поддерживает детерминированную очистку асинхронных генераторов, когда они завершаются досрочно из-за break или исключения. Например:

from contextlib import aclosing

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

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

Новое в версии 3.10.

contextlib.nullcontext(enter_result=None)¶

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

def myfunction(arg, ignore_exceptions=False):
    if ignore_exceptions:
        # Используйте suppress, чтобы игнорировать все исключения.
        cm = contextlib.suppress(Exception)
    else:
        # Не игнорируйте никакие исключения, cm не оказывает эффекта.
        cm = contextlib.nullcontext()
    with cm:
        # Выполнить действие

Пример использования enter_result:

def process_file(file_or_path):
    if isinstance(file_or_path, str):
        # Если строка, открыть файл
        cm = open(file_or_path)
    else:
        # Вызывающий отвечает за закрытие файла
        cm = nullcontext(file_or_path)

    with cm as file:
        # Выполнить обработку файла

Его также можно использовать в качестве замены асинхронных менеджеров контекста:

async def send_http(session=None):
    if not session:
        # Если нет http-сессии, создать её с помощью aiohttp
        cm = aiohttp.ClientSession()
    else:
        # Вызывающий отвечает за закрытие сессии
        cm = nullcontext(session)

    async with cm as session:
        # Отправить http-запросы с помощью сессии

Добавлено в версии 3.7.

Изменено в версии 3.10: добавлена поддержка асинхронного менеджера контекста.

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. Заменяющий поток данных возвращается из метода __enter__ и поэтому доступен как цель оператора with:

with redirect_stdout(io.StringIO()) as 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.

contextlib.chdir(path)¶

Менеджер контекста, небезопасный для параллельного использования, для смены текущей рабочей директории. Поскольку это меняет глобальное состояние – рабочую директорию, он не подходит для использования в большинстве многопоточных или асинхронных контекстов. Также он не подходит для нелинейного выполнения кода, например, генераторов, где выполнение программы временно приостанавливается – если только это не требуется явно, не следует выполнять yield, пока этот менеджер контекста активен.

Это простая обёртка вокруг chdir(): она меняет текущую рабочую директорию при входе и восстанавливает предыдущую при выходе.

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

Новое в версии 3.11.

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():
        # Выполнить действия

ContextDecorator позволяет вместо этого писать:

@cm()
def f():
    # Выполнить действия

Становится ясно, что cm применяется ко всей функции, а не только к её части (и экономия уровня отступа тоже приятна).

Существующие менеджеры контекста, уже имеющие базовый класс, можно расширить, используя ContextDecorator в качестве примеси (mixin):

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.AsyncContextDecorator¶

Аналогично ContextDecorator, но только для асинхронных функций.

Пример 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¶

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

Например, набор файлов можно легко обработать в одном операторе with следующим образом:

with ExitStack() as stack:
    files = [stack.enter_context(open(fname)) for fname in filenames]
    # Все открытые файлы будут автоматически закрыты в конце
    # оператор with, даже если попытки открыть файлы, идущие далее
    # в списке, вызывают исключение

Метод __enter__() возвращает экземпляр ExitStack и не выполняет никаких дополнительных операций.

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

Такая стековая модель используется, чтобы менеджеры контекста, получающие свои ресурсы в методе __init__ (например, файловые объекты), могли обрабатываться корректно.

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

Это относительно низкоуровневый API, который заботится о деталях корректной размотки стека колбэков выхода. Он предоставляет подходящую основу для менеджеров контекста более высокого уровня, которые манипулируют стеком выхода специфичным для приложения образом.

Новое в версии 3.3.

enter_context(cm)¶

Входит в новый менеджер контекста и добавляет его метод __exit__() в стек колбэков. Возвращаемое значение – результат собственного метода __enter__() менеджера контекста.

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

Изменено в версии 3.11: Возбуждает TypeError вместо AttributeError, если cm не является менеджером контекста.

push(exit)¶

Добавляет метод __exit__() менеджера контекста в стек колбэков.

Поскольку __enter__ не вызывается, этот метод можно использовать для покрытия части реализации __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]
    # Сохраните метод close, но пока не вызывайте его.
    close_files = stack.pop_all().close
    # Если открытие любого файла завершается неудачей, все ранее открытые файлы будут
    # закрыты автоматически. Если все файлы открыты успешно,
    # они останутся открытыми даже после завершения оператора with.
    # close_files() затем можно вызвать явно, чтобы закрыть их все.

close()¶

Немедленно разворачивает стек колбэков, вызывая колбэки в порядке, обратном регистрации. Для любых зарегистрированных менеджеров контекста и колбэков выхода переданные аргументы будут указывать, что исключения не произошло.

class contextlib.AsyncExitStack¶

Асинхронный менеджер контекста, аналогичный ExitStack, который поддерживает объединение как синхронных, так и асинхронных менеджеров контекста, а также имеет корутины для логики очистки.

Метод close() не реализован; вместо него следует использовать aclose().

coroutine enter_async_context(cm)¶

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

Изменено в версии 3.11: Вызывает TypeError вместо AttributeError, если cm не является асинхронным менеджером контекста.

push_async_exit(exit)¶

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

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

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

coroutine aclose()¶

Аналогично ExitStack.close(), но корректно обрабатывает ожидаемые объекты.

Продолжая пример для asynccontextmanager():

async with AsyncExitStack() as stack:
    connections = [await stack.enter_async_context(get_connection())
        for i in range(5)]
    # Все открытые соединения будут автоматически освобождены в конце
    # оператора async with, даже если попытки открыть соединение
    # далее в списке вызывают исключение.

Добавлено в версии 3.7.

Примеры и рецепты¶Examples and Recipes

В этом разделе описаны некоторые примеры и рецепты для эффективного использования инструментов, предоставляемых 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)
    # Выполните операции, использующие полученные ресурсы

stack = ExitStack()
try:
    x = stack.enter_context(cm)
except Exception:
    # обработать исключение __enter__
else:
    with stack:
        # Обработка нормального случая

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
            # Проверка прошла успешно и не вызвала исключение
            # Соответственно, мы хотим сохранить ресурс и передать его
            # обратно вызывающему коду
            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):
        # Нам не нужно дублировать логику освобождения ресурсов
        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

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

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

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

Менеджеры контекста, созданные с помощью contextmanager(), также являются одноразовыми менеджерами контекста и будут сообщать об ошибке, если базовый генератор не сможет выполнить yield при попытке использовать их второй раз:

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