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

Декорируемая функция при вызове должна возвращать итератор-генератор. Этот итератор должен выдавать ровно одно значение, которое будет привязано к целевым переменным в предложении as оператора 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.

Контекстные менеджеры, определённые с помощью 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():
    # ... async code ...

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

Примечание

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

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

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: Добавлена поддержка асинхронного контекстного менеджера.

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

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

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

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

Изменено в версии 3.12: Теперь suppress поддерживает подавление исключений, возникших в составе BaseExceptionGroup.

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():
        # 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.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]
    # 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

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

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

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

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

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

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

enter_context(cm)

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

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

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

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]
    # 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().

async enter_async_context(cm)

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

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

push_async_exit(exit)

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

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

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

async aclose()

Аналогичен ExitStack.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.

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