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

28.6.1. Утилиты¶Utilities

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

@contextlib.contextmanager¶

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

Простой пример (это не рекомендуется как реальный способ генерации HTML!):

from contextlib import contextmanager

@contextmanager
def tag(name):
    print("<%s>" % name)
    yield
    print("</%s>" % name)

>>> with tag("h1"):
...    print("foo")
...
<h1>
foo
</h1>

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

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

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

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

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.

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 class):

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]
    # Все открытые файлы будут автоматически закрыты в конце
    # оператор with, даже если попытки открыть файлы, идущие далее
    # в списке, вызывают исключение

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

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

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

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

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

enter_context(cm)¶

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

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

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

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

28.6.2. Примеры и рецепты¶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)
    # Выполните операции, использующие полученные ресурсы

def debug_trace(details):
    if __debug__:
        return TraceContext(details)
    # В релизном режиме никаких особых действий с контекстом не выполняется
    return ExitStack()

with debug_trace():
    # В режиме отладки набор трассируется, в остальных случаях работает обычным образом

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

from contextlib import contextmanager, ExitStack

class ResourceManager:

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

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: {}'.format(name))

    def __exit__(self, exc_type, exc, exc_tb):
        logging.info('Exiting: {}'.format(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()