Модуль Python в неделю | contextlib

Адрес столбца:Один модуль Python в неделю

В то же время, вы также можете обратить внимание на мой публичный аккаунт WeChat AlwaysBeta, вас ждет более интересный контент.

Утилиты для создания и использования контекстных менеджеров.

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

Context Manager API

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

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

withоператор включает диспетчер контекста, а API включает два метода: запуск, когда поток выполнения входит во внутренний блок кода__enter__()метод, который возвращает объект, который будет использоваться в контексте. Когда поток выполнения покидаетwithблок, контекстный менеджер__exit__()метод очистки любых используемых ресурсов.

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')
    
# output
# __init__()
# __enter__()
# Doing work in the context
# __exit__()

Объединение контекстных менеджеров иwithЭто более краткое заявлениеtry:finallyблокировать, даже если выдается исключение, диспетчер контекста__exit__()метод.

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

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()
    
# output
# Context.__init__()
# Context.__enter__()
# WithinContext.__init__(<__main__.Context object at 0x101f046d8>)
# WithinContext.do_something()
# Context.__exit__()
# WithinContext.__del__

значение, связанное с переменнойcвозвращается__enter__()объект, который не должен бытьContextсуществуетwithЭкземпляр, созданный в операторе.

__exit__()Метод получает содержитwithПараметр для сведений о любых исключениях, созданных в блоке.

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')
    
# output
# __init__(True)
# __enter__()
# __exit__()
#   exc_type = <class 'RuntimeError'>
#   exc_val  = error message handled
#   exc_tb   = <traceback object at 0x101c94948>
# 
# __init__(False)
# __enter__()
# __exit__()
#   exc_type = <class 'RuntimeError'>
#   exc_val  = error message propagated
#   exc_tb   = <traceback object at 0x101c94948>
# Traceback (most recent call last):
#   File "contextlib_api_error.py", line 34, in <module>
#     raise RuntimeError('error message propagated')
# RuntimeError: error message propagated

Если диспетчер контекста может обработать исключение,__exit__()Значение true должно быть возвращено, чтобы указать, что исключение не нужно распространять, false вызовет__exit__()Повторно выдать исключение после возврата.

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

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

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

# output
# __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__()Возвращаемое значение недоступно внутри декорированной функции, что аналогично использованиюwithиasПараметры, передаваемые декорированной функции, предоставляются обычным образом.

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

используя__enter__()и__exit__()Традиционный способ написания класса для создания менеджера контекста не сложен. Однако иногда выписывать все не обязательно для какого-то тривиального контекста. В этих случаях используйтеcontextmanager()Декоратор превращает функцию-генератор в менеджер контекста.

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')
    
# output
# 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 <module>
#     raise ValueError('this exception is not handled')
# ValueError: this exception is not handled

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

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

@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'))

# output
# 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
# <module>
#     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

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

закрыть открытую ручку

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

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)
    
# output
# 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

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

игнорировать исключение

Наиболее распространенный способ игнорировать исключения — использовать блоки операторов.try:except, то в заявленииexceptтолько вpass.

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

# output
# trying non-idempotent operation
# done

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

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

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

# output
# trying non-idempotent operation
# done

В этой обновленной версии исключения полностью удалены.

перенаправить поток вывода

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

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

# output
# (stdout) A: 5
# (stderr) A: 5

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

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

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

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

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

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

Есть несколько способов заполненияExitStack. Этот пример дляenter_context()Добавьте новый контекстный менеджер в стек.

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

# output
# 0 entering
# 1 entering
# inside context
# 1 exiting
# 0 exiting

enter_context()первый звонок__enter__()диспетчер контекста, затем__exit__()Метод зарегистрирован как обратный вызов, который будет вызываться, когда стек втягивается.

Менеджер контекста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Также поддерживаются произвольные обратные вызовы для закрытия контекста, что упрощает очистку ресурсов, не контролируемых менеджером контекста.

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')
    
# output
# closing callback((), {'arg3': 'val3'})
# closing callback(('arg1', 'arg2'), {})

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

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

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))
    
# output
# closing callback((), {'arg3': 'val3'})
# closing callback(('arg1', 'arg2'), {})
# ERROR: thrown error

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

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

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')
    
# output
# within the context
# inline_cleanup()
# local_resource = 'resource created in context'

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

неполный стек

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

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

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')
        
# output
# 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

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

Связанные документы:

Дешевый Мо Corruption.com/3/context…