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. Даже если произойдёт ошибка, 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:
        # 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: Вызывает 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]
    # 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().

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(), но правильно обрабатывает 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