Рубрики
Без рубрики

contextlib – Утилиты диспетчера контекста

Автор оригинала: Doug Hellmann.

Цель:

Утилиты для создания и работы с менеджерами контекста.

Модуль contextlib содержит утилиты для работы с менеджерами контекста и оператор with .

API диспетчера контекста

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

contextlib_file.py

with open('/tmp/pymotw.txt', 'wt') as f:
    f.write('contents go here')
# file is automatically closed

Диспетчер контекста включается оператором with , а API включает два метода. Метод __enter __ () запускается, когда поток выполнения входит в блок кода внутри with . Он возвращает объект, который будет использоваться в контексте. Когда поток выполнения покидает блок with , вызывается метод __exit __ () диспетчера контекста для очистки всех используемых ресурсов.

contextlib_api.py

class Context:

    def __init__(self):
        print('__init__()')

    def __enter__(self):
        print('__enter__()')
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        print('__exit__()')


with Context():
    print('Doing work in the context')

Объединение диспетчера контекста и оператора with – это более компактный способ записи блока try: finally , поскольку метод __exit __ () диспетчера контекста вызывается всегда, даже если возникает исключение.

$ python3 contextlib_api.py

__init__()
__enter__()
Doing work in the context
__exit__()

Метод __enter __ () может возвращать любой объект, который должен быть связан с именем, указанным в предложении as оператора with . В этом примере Context возвращает объект, который использует открытый контекст.

contextlib_api_other_object.py

class WithinContext:

    def __init__(self, context):
        print('WithinContext.__init__({})'.format(context))

    def do_something(self):
        print('WithinContext.do_something()')

    def __del__(self):
        print('WithinContext.__del__')


class Context:

    def __init__(self):
        print('Context.__init__()')

    def __enter__(self):
        print('Context.__enter__()')
        return WithinContext(self)

    def __exit__(self, exc_type, exc_val, exc_tb):
        print('Context.__exit__()')


with Context() as c:
    c.do_something()

Значение, связанное с переменной c , является объектом, возвращаемым __enter __ () , который не обязательно является экземпляром Context , созданным в с оператором .

$ python3 contextlib_api_other_object.py

Context.__init__()
Context.__enter__()
WithinContext.__init__(<__main__.Context object at 0x101f046d8>)
WithinContext.do_something()
Context.__exit__()
WithinContext.__del__

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

contextlib_api_error.py

class Context:

    def __init__(self, handle_error):
        print('__init__({})'.format(handle_error))
        self.handle_error  handle_error

    def __enter__(self):
        print('__enter__()')
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        print('__exit__()')
        print('  exc_type =', exc_type)
        print('  exc_val  =', exc_val)
        print('  exc_tb   =', exc_tb)
        return self.handle_error


with Context(True):
    raise RuntimeError('error message handled')

print()

with Context(False):
    raise RuntimeError('error message propagated')

Если диспетчер контекста может обработать исключение, __exit __ () должен вернуть истинное значение, чтобы указать, что исключение не нужно распространять. Возврат false вызывает повторное возникновение исключения после возврата __exit __ () .

$ python3 contextlib_api_error.py

__init__(True)
__enter__()
__exit__()
  exc_type = 
  exc_val  = error message handled
  exc_tb   = 

__init__(False)
__enter__()
__exit__()
  exc_type = 
  exc_val  = error message propagated
  exc_tb   = 
Traceback (most recent call last):
  File "contextlib_api_error.py", line 34, in 
    raise RuntimeError('error message propagated')
RuntimeError: error message propagated

Менеджеры контекста как декораторы функций

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

contextlib_decorator.py

import contextlib


class Context(contextlib.ContextDecorator):

    def __init__(self, how_used):
        self.how_used  how_used
        print('__init__({})'.format(how_used))

    def __enter__(self):
        print('__enter__({})'.format(self.how_used))
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        print('__exit__({})'.format(self.how_used))


@Context('as decorator')
def func(message):
    print(message)


print()
with Context('as context manager'):
    print('Doing work in the context')

print()
func('Doing work in the wrapped function')

Одно различие с использованием диспетчера контекста в качестве декоратора заключается в том, что значение, возвращаемое __enter __ () , недоступно внутри декорируемой функции, в отличие от использования with и как . Аргументы, передаваемые декорируемой функции, доступны обычным образом.

$ python3 contextlib_decorator.py

__init__(as decorator)

__init__(as context manager)
__enter__(as context manager)
Doing work in the context
__exit__(as context manager)

__enter__(as decorator)
Doing work in the wrapped function
__exit__(as decorator)

От генератора к контекстному менеджеру

Создать менеджеры контекста традиционным способом, написав класс с методами __enter __ () и __exit __ () , не сложно. Но иногда полное написание всего – это лишние накладные расходы из-за банального контекста. В таких ситуациях используйте декоратор contextmanager () для преобразования функции генератора в диспетчер контекста.

contextlib_contextmanager.py

import contextlib


@contextlib.contextmanager
def make_context():
    print('  entering')
    try:
        yield {}
    except RuntimeError as err:
        print('  ERROR:', err)
    finally:
        print('  exiting')


print('Normal:')
with make_context() as value:
    print('  inside with statement:', value)

print('\nHandled error:')
with make_context() as value:
    raise RuntimeError('showing example of handling an error')

print('\nUnhandled error:')
with make_context() as value:
    raise ValueError('this exception is not handled')

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

$ python3 contextlib_contextmanager.py

Normal:
  entering
  inside with statement: {}
  exiting

Handled error:
  entering
  ERROR: showing example of handling an error
  exiting

Unhandled error:
  entering
  exiting
Traceback (most recent call last):
  File "contextlib_contextmanager.py", line 33, in 
    raise ValueError('this exception is not handled')
ValueError: this exception is not handled

Диспетчер контекста, возвращаемый функцией contextmanager () , является производным от ContextDecorator , поэтому он также работает как декоратор функций.

contextlib_contextmanager_decorator.py

import contextlib


@contextlib.contextmanager
def make_context():
    print('  entering')
    try:
        # Yield control, but not a value, because any value
        # yielded is not available when the context manager
        # is used as a decorator.
        yield
    except RuntimeError as err:
        print('  ERROR:', err)
    finally:
        print('  exiting')


@make_context()
def normal():
    print('  inside with statement')


@make_context()
def throw_error(err):
    raise err


print('Normal:')
normal()

print('\nHandled error:')
throw_error(RuntimeError('showing example of handling an error'))

print('\nUnhandled error:')
throw_error(ValueError('this exception is not handled'))

Как и в приведенном выше примере ContextDecorator , когда диспетчер контекста используется в качестве декоратора, значение, выдаваемое генератором, недоступно внутри декорируемой функции. Аргументы, переданные в декорированную функцию, по-прежнему доступны, как показано throw_error () в этом примере.

$ python3 contextlib_contextmanager_decorator.py

Normal:
  entering
  inside with statement
  exiting

Handled error:
  entering
  ERROR: showing example of handling an error
  exiting

Unhandled error:
  entering
  exiting
Traceback (most recent call last):
  File "contextlib_contextmanager_decorator.py", line 43, in

    throw_error(ValueError('this exception is not handled'))
  File ".../lib/python3.7/contextlib.py", line 74, in inner
    return func(*args, **kwds)
  File "contextlib_contextmanager_decorator.py", line 33, in
throw_error
    raise err
ValueError: this exception is not handled

Закрытие открытых ручек

Класс file напрямую поддерживает API диспетчера контекста, но некоторые другие объекты, представляющие открытые дескрипторы, не поддерживают. Пример, приведенный в документации стандартной библиотеки для contextlib , – это объект, возвращаемый из urllib.urlopen () . Существуют и другие устаревшие классы, которые используют метод close () , но не поддерживают API диспетчера контекста. Чтобы гарантировать, что дескриптор закрыт, используйте close () , чтобы создать для него диспетчер контекста.

contextlib_closing.py

import contextlib


class Door:

    def __init__(self):
        print('  __init__()')
        self.status  'open'

    def close(self):
        print('  close()')
        self.status  'closed'


print('Normal Example:')
with contextlib.closing(Door()) as door:
    print('  inside with statement: {}'.format(door.status))
print('  outside with statement: {}'.format(door.status))

print('\nError handling example:')
try:
    with contextlib.closing(Door()) as door:
        print('  raising from inside with statement')
        raise RuntimeError('error message')
except Exception as err:
    print('  Had an error:', err)

Дескриптор закрывается независимо от того, есть ли ошибка в блоке with .

$ python3 contextlib_closing.py

Normal Example:
  __init__()
  inside with statement: open
  close()
  outside with statement: closed

Error handling example:
  __init__()
  raising from inside with statement
  close()
  Had an error: error message

Игнорирование исключений

Часто бывает полезно игнорировать исключения, создаваемые библиотеками, поскольку ошибка указывает на то, что желаемое состояние уже достигнуто, или ее можно игнорировать в противном случае. Наиболее распространенный способ игнорировать исключения – использовать оператор try: except , содержащий только оператор pass в блоке except .

contextlib_ignore_error.py

import contextlib


class NonFatalError(Exception):
    pass


def non_idempotent_operation():
    raise NonFatalError(
        'The operation failed because of existing state'
    )


try:
    print('trying non-idempotent operation')
    non_idempotent_operation()
    print('succeeded!')
except NonFatalError:
    pass

print('done')

В этом случае операция не выполняется, и ошибка игнорируется.

$ python3 contextlib_ignore_error.py

trying non-idempotent operation
done

Форму try: except можно заменить на contextlib.suppress () , чтобы более явно подавить класс исключений, происходящих где-либо в блоке with .

contextlib_suppress.py

import contextlib


class NonFatalError(Exception):
    pass


def non_idempotent_operation():
    raise NonFatalError(
        'The operation failed because of existing state'
    )


with contextlib.suppress(NonFatalError):
    print('trying non-idempotent operation')
    non_idempotent_operation()
    print('succeeded!')

print('done')

В этой обновленной версии исключение полностью исключено.

$ python3 contextlib_suppress.py

trying non-idempotent operation
done

Перенаправление выходных потоков

Плохо спроектированный код библиотеки может писать напрямую в sys.stdout или sys.stderr , без предоставления аргументов для настройки различных мест назначения вывода. Менеджеры контекста redirect_stdout () и redirect_stderr () могут использоваться для захвата вывода от таких функций, для которых источник не может быть изменен для принятия нового аргумента вывода.

contextlib_redirect.py

from contextlib import redirect_stdout, redirect_stderr
import io
import sys


def misbehaving_function(a):
    sys.stdout.write('(stdout) A: {!r}\n'.format(a))
    sys.stderr.write('(stderr) A: {!r}\n'.format(a))


capture  io.StringIO()
with redirect_stdout(capture), redirect_stderr(capture):
    misbehaving_function(5)

print(capture.getvalue())

В этом примере misbehaving_function () записывает в оба stdout и stderr , но два диспетчера контекста отправляют этот вывод в один и тот же io .StringIO , где он сохраняется для дальнейшего использования.

$ python3 contextlib_redirect.py

(stdout) A: 5
(stderr) A: 5

Примечание

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

Стеки диспетчера динамического контекста

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

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

Составление менеджеров контекста

Есть несколько способов заполнить ExitStack . В этом примере используется enter_context () для добавления нового диспетчера контекста в стек.

contextlib_exitstack_enter_context.py

import contextlib


@contextlib.contextmanager
def make_context(i):
    print('{} entering'.format(i))
    yield {}
    print('{} exiting'.format(i))


def variable_stack(n, msg):
    with contextlib.ExitStack() as stack:
        for i in range(n):
            stack.enter_context(make_context(i))
        print(msg)


variable_stack(2, 'inside context')

enter_context () сначала вызывает __enter __ () в диспетчере контекста, а затем регистрирует свой метод __exit __ () как обратный вызов, который будет вызываться как стек отменен.

$ python3 contextlib_exitstack_enter_context.py

0 entering
1 entering
inside context
1 exiting
0 exiting

Менеджеры контекста, данные для ExitStack , обрабатываются так, как если бы они были в серии вложенных операторов with . Ошибки, которые происходят в любом месте контекста, распространяются через обычную обработку ошибок диспетчеров контекста. Эти классы диспетчера контекста иллюстрируют способ распространения ошибок.

contextlib_context_managers.py

import contextlib


class Tracker:
    "Base class for noisy context managers."

    def __init__(self, i):
        self.i  i

    def msg(self, s):
        print('  {}({}): {}'.format(
            self.__class__.__name__, self.i, s))

    def __enter__(self):
        self.msg('entering')


class HandleError(Tracker):
    "If an exception is received, treat it as handled."

    def __exit__(self, *exc_details):
        received_exc  exc_details[1] is not None
        if received_exc:
            self.msg('handling exception {!r}'.format(
                exc_details[1]))
        self.msg('exiting {}'.format(received_exc))
        # Return Boolean value indicating whether the exception
        # was handled.
        return received_exc


class PassError(Tracker):
    "If an exception is received, propagate it."

    def __exit__(self, *exc_details):
        received_exc  exc_details[1] is not None
        if received_exc:
            self.msg('passing exception {!r}'.format(
                exc_details[1]))
        self.msg('exiting')
        # Return False, indicating any exception was not handled.
        return False


class ErrorOnExit(Tracker):
    "Cause an exception."

    def __exit__(self, *exc_details):
        self.msg('throwing error')
        raise RuntimeError('from {}'.format(self.i))


class ErrorOnEnter(Tracker):
    "Cause an exception."

    def __enter__(self):
        self.msg('throwing error on enter')
        raise RuntimeError('from {}'.format(self.i))

    def __exit__(self, *exc_info):
        self.msg('exiting')

Примеры с использованием этих классов основаны на variable_stack () , который использует переданные диспетчеры контекста для создания ExitStack , создавая общий контекст один за другим. В приведенных ниже примерах передаются различные менеджеры контекста для изучения поведения обработки ошибок. Во-первых, нормальный случай без исключений.

print('No errors:')
variable_stack([
    HandleError(1),
    PassError(2),
])

Затем пример обработки исключений в диспетчерах контекста в конце стека, в котором все открытые контексты закрываются по мере разматывания стека.

print('\nError at the end of the context stack:')
variable_stack([
    HandleError(1),
    HandleError(2),
    ErrorOnExit(3),
])

Затем пример обработки исключений в диспетчерах контекста в середине стека, в котором ошибка не возникает, пока некоторые контексты уже не закрыты, поэтому эти контексты не видят ошибку.

print('\nError in the middle of the context stack:')
variable_stack([
    HandleError(1),
    PassError(2),
    ErrorOnExit(3),
    HandleError(4),
])

Наконец, пример исключения, которое остается необработанным и распространяется до вызывающего кода.

try:
    print('\nError ignored:')
    variable_stack([
        PassError(1),
        ErrorOnExit(2),
    ])
except RuntimeError:
    print('error handled outside of context')

Если какой-либо диспетчер контекста в стеке получает исключение и возвращает значение True , он предотвращает распространение этого исключения на любые другие диспетчеры контекста.

$ python3 contextlib_exitstack_enter_context_errors.py

No errors:
  HandleError(1): entering
  PassError(2): entering
  PassError(2): exiting
  HandleError(1): exiting False
  outside of stack, any errors were handled

Error at the end of the context stack:
  HandleError(1): entering
  HandleError(2): entering
  ErrorOnExit(3): entering
  ErrorOnExit(3): throwing error
  HandleError(2): handling exception RuntimeError('from 3')
  HandleError(2): exiting True
  HandleError(1): exiting False
  outside of stack, any errors were handled

Error in the middle of the context stack:
  HandleError(1): entering
  PassError(2): entering
  ErrorOnExit(3): entering
  HandleError(4): entering
  HandleError(4): exiting False
  ErrorOnExit(3): throwing error
  PassError(2): passing exception RuntimeError('from 3')
  PassError(2): exiting
  HandleError(1): handling exception RuntimeError('from 3')
  HandleError(1): exiting True
  outside of stack, any errors were handled

Error ignored:
  PassError(1): entering
  ErrorOnExit(2): entering
  ErrorOnExit(2): throwing error
  PassError(1): passing exception RuntimeError('from 2')
  PassError(1): exiting
error handled outside of context

Обратные вызовы произвольного контекста

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

contextlib_exitstack_callbacks.py

import contextlib


def callback(*args, **kwds):
    print('closing callback({}, {})'.format(args, kwds))


with contextlib.ExitStack() as stack:
    stack.callback(callback, 'arg1', 'arg2')
    stack.callback(callback, arg3'val3')

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

$ python3 contextlib_exitstack_callbacks.py

closing callback((), {'arg3': 'val3'})
closing callback(('arg1', 'arg2'), {})

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

contextlib_exitstack_callbacks_error.py

import contextlib


def callback(*args, **kwds):
    print('closing callback({}, {})'.format(args, kwds))


try:
    with contextlib.ExitStack() as stack:
        stack.callback(callback, 'arg1', 'arg2')
        stack.callback(callback, arg3'val3')
        raise RuntimeError('thrown error')
except RuntimeError as err:
    print('ERROR: {}'.format(err))

Поскольку они не имеют доступа к ошибке, обратные вызовы не могут подавить распространение исключений через остальную часть стека диспетчеров контекста.

$ python3 contextlib_exitstack_callbacks_error.py

closing callback((), {'arg3': 'val3'})
closing callback(('arg1', 'arg2'), {})
ERROR: thrown error

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

contextlib_exitstack_callbacks_decorator.py

import contextlib


with contextlib.ExitStack() as stack:

    @stack.callback
    def inline_cleanup():
        print('inline_cleanup()')
        print('local_resource = {!r}'.format(local_resource))

    local_resource  'resource created in context'
    print('within the context')

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

$ python3 contextlib_exitstack_callbacks_decorator.py

within the context
inline_cleanup()
local_resource = 'resource created in context'

Частичные стеки

Иногда при построении сложных контекстов полезно иметь возможность прервать операцию, если контекст не может быть полностью сконструирован, но отложить очистку всех ресурсов на более позднее время, если все они могут быть правильно настроены. Например, если для операции требуется несколько долговечных сетевых подключений, может быть лучше не запускать операцию, если одно подключение не удается. Однако, если все соединения могут быть открыты, они должны оставаться открытыми дольше, чем время действия одного диспетчера контекста. В этом сценарии можно использовать метод pop_all () из ExitStack .

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

contextlib_exitstack_pop_all.py

import contextlib

from contextlib_context_managers import *


def variable_stack(contexts):
    with contextlib.ExitStack() as stack:
        for c in contexts:
            stack.enter_context(c)
        # Return the close() method of a new stack as a clean-up
        # function.
        return stack.pop_all().close
    # Explicitly return None, indicating that the ExitStack could
    # not be initialized cleanly but that cleanup has already
    # occurred.
    return None


print('No errors:')
cleaner  variable_stack([
    HandleError(1),
    HandleError(2),
])
cleaner()

print('\nHandled error building context manager stack:')
try:
    cleaner  variable_stack([
        HandleError(1),
        ErrorOnEnter(2),
    ])
except RuntimeError as err:
    print('caught error {}'.format(err))
else:
    if cleaner is not None:
        cleaner()
    else:
        print('no cleaner returned')

print('\nUnhandled error building context manager stack:')
try:
    cleaner  variable_stack([
        PassError(1),
        ErrorOnEnter(2),
    ])
except RuntimeError as err:
    print('caught error {}'.format(err))
else:
    if cleaner is not None:
        cleaner()
    else:
        print('no cleaner returned')

В этом примере используются те же классы диспетчера контекста, которые были определены ранее, с той разницей, что ErrorOnEnter выдает ошибку в __enter __ () вместо __exit __ () . Внутри variable_stack () , если все контексты введены без ошибок, возвращается метод close () нового ExitStack . Если происходит обработанная ошибка, variable_stack () возвращает None , чтобы указать, что работа по очистке уже выполнена. И если возникает необработанная ошибка, неполный стек очищается, и ошибка распространяется.

$ python3 contextlib_exitstack_pop_all.py

No errors:
  HandleError(1): entering
  HandleError(2): entering
  HandleError(2): exiting False
  HandleError(1): exiting False

Handled error building context manager stack:
  HandleError(1): entering
  ErrorOnEnter(2): throwing error on enter
  HandleError(1): handling exception RuntimeError('from 2')
  HandleError(1): exiting True
no cleaner returned

Unhandled error building context manager stack:
  PassError(1): entering
  ErrorOnEnter(2): throwing error on enter
  PassError(1): passing exception RuntimeError('from 2')
  PassError(1): exiting
caught error from 2

Смотрите также