contextlib — Utilidades de gestión de contexto

Propósito:Utilidades para crear y trabajar con gestores de contexto.

El módulo contextlib contiene utilidades para trabajar con gestores de contexto y la declaración with.

Interfaz de gestor de contexto

Un gestor de contexto es responsable de un recurso dentro de un bloque de código, posiblemente creandolo cuando se ingrese el bloque y luego limpiandolo después de salir del bloque. Por ejemplo, los archivos admiten la interfaz de gestión de contexto para que sea más fácil asegurarse de que estén cerrados después que de todo el leer o escribir está hecho.

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

Un gestor de contexto es habilitado por la declaración with, y la interfaz implica dos métodos. El método __enter__() se ejecuta cuando el flujo de ejecución ingresa al bloque de código dentro del with. Éste devuelve un objeto para ser utilizado dentro del contexto. Cuando el flujo de ejecución deja el bloque with, el método __exit__() del gestor de contexto se ejecuta para limpiar los recursos que se utilizan.

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

La combinación de un gestor de contexto y la declaración with es una forma más compacta de escribir un bloque try:finally, ya que siempre se llama al método __exit__() del gestor de contexto, incluso si se plantea una excepción

$ python3 contextlib_api.py

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

El método __enter__() puede devolver cualquier objeto para ser asociado con un nombre especificado en la cláusula as de la declaración with. En este ejemplo, el Context devuelve un objeto que usa el contexto abierto.

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

El valor asociado con la variable c es el objeto devuelto por __enter__(), que no es necesariamente la instancia Context creada en la declaración with.

$ python3 contextlib_api_other_object.py

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

El método __exit__() recibe argumentos que contienen detalles de cualquier excepción planteada en el bloque 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')

Si el gestor de contexto puede manejar la excepción, __exit__() debe devolver un valor verdadero para indicar que la excepción no necesita ser propagada. Devolver falso hace que la excepción sea propagada después de que __exit__() regresa.

$ python3 contextlib_api_error.py

__init__(True)
__enter__()
__exit__()
  exc_type = <class 'RuntimeError'>
  exc_val  = error message handled
  exc_tb   = <traceback object at 0x101b5cc88>

__init__(False)
__enter__()
__exit__()
  exc_type = <class 'RuntimeError'>
  exc_val  = error message propagated
  exc_tb   = <traceback object at 0x101b5cc88>
Traceback (most recent call last):
  File "contextlib_api_error.py", line 34, in <module>
    raise RuntimeError('error message propagated')
RuntimeError: error message propagated

Gestores de contexto como decoradores de funciones

La clase ContextDecorator agrega soporte a clases normales de gestores de contexto para que puedan ser utilizadas como decoradores de funciones, así como gestores de contexto.

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

Una diferencia con el uso del gestor de contexto como decorador es que el valor devuelto por __enter__() no está disponible dentro de la función que es decorada, a diferencia de cuando se usa with y as. Los argumentos pasados a la función decorada están disponibles de la manera habitual.

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

Del generador al gestor de contexto

Crear gestores de contexto de la manera tradicional, escribiendo una clase con métods __enter__() y __exit__(), no es difícil. Pero a veces escribir todo es sobrecarga extra para un poco de contexto trivial. En ese tipo de situaciones, usa el decorador contextmanager() para convertir una función generador en un gestor de contexto.

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

El generador debe inicializar el contexto, ceder exactamente una vez, luego limpia el contexto. El valor obtenido, si lo hay, está vinculado a la variable en la cláusula as de la declaración with. Las excepciones dentro del bloque with son propagadas dentro del generador, para que puedan ser manejadas allí.

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

El gestor de contexto devuelto por contextmanager() se deriva de ContextDecorator, por lo que también funciona como un decorador de funciones.

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

Como en el ejemplo anterior de ContextDecorator, cuando el gestor de contexto se utiliza como decorador el valor producido por el generador no está disponible dentro de la función que se está decorando. Argumentos pasados a la función decorada todavía están disponible, como lo demuestra throw_error() en este ejemplo.

$ 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
<module>
    throw_error(ValueError('this exception is not handled'))
  File ".../lib/python3.5/contextlib.py", line 30, in inner
    return func(*args, **kwds)
  File "contextlib_contextmanager_decorator.py", line 33, in
throw_error
    raise err
ValueError: this exception is not handled

Cerrando manejadores abiertos

La clase file admite la interfaz de gestor de contexto directamente, pero algunos otros objetos que representan controladores abiertos no. El ejemplo dado en la documentación de la biblioteca estándar para contextlib es el objeto devuelto por urllib.urlopen(). Hay otras clases heredadas que usan un método close() pero no admiten interfaz de gestor de contexto. Para asegurarse de que un manejador esté cerrado, usa closing() para crear un gestor de contexto para él.

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)

El manejador está cerrado si hay un error en el bloque with o no.

$ 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

Ignorando excepciones

Con frecuencia es útil ignorar las excepciones planteadas por las bibliotecas, porque el error indica que el estado deseado ya ha sido logrado, o de lo contrario puede ser ignorado. La forma más común de ignorar excepciones es con una declaración try:except con solo un pass en el bloque 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')

En este caso, la operación falla y el error es ignorado.

$ python3 contextlib_ignore_error.py

trying non-idempotent operation
done

La forma try: except puede ser reemplazada por contextlib.suppress() para suprimir más explícitamente una clase de excepciones que suceden en cualquier lugar en el bloque 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')

En esta versión actualizada, la excepción se descarta por completo.

$ python3 contextlib_suppress.py

trying non-idempotent operation
done

Redirigiendo flujos de salida

El código de la biblioteca mal diseñado puede escribir directamente en sys.stdout o sys.stderr, sin proporcionar argumentos para configurar diferentes destinos de salida. Los gestores de contexto redirect_stdout() y redirect_stderr() se pueden usar para capturar resultados de funciones como esta, para las cuales la fuente no se puede cambiar para aceptar un nuevo argumento de salida.

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

En este ejemplo, misbehaving_function() escribe en ambos stdout y stderr, pero los dos gestores de contexto envían esa salida a la misma instancia io.StringIO donde se guarda para ser utilizada más tarde.

$ python3 contextlib_redirect.py

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

Nota

Tanto redirect_stdout() como redirect_stderr() modifican el estado global reemplazando objetos en el módulo sys, y deben usarse con cuidado. Las funciones no son seguras para subprocesos y pueden interferir con otras operaciones que esperan el flujo estándar para conectarlo a dispositivos terminales.

Pilas de gestores de contexto dinámicas

La mayoría de los gestores de contexto operan en un objeto a la vez, como un archivo único o manejador de base de datos. En estos casos, el objeto es conocido con aticipación y el código que utiliza el gestor de contexto se puede construir alrededor de ese único objeto. En otros casos, un programa puede necesitar crear un número desconocido de objetos en un contexto, mientras que todos quieren que se limpien cuando el flujo de control sale del contexto. ExitStack fue creado para manejar estos casos más dinámicos.

Una instancia de ExitStack mantiene una estructura de datos de pila de llamadas de limpieza. Las devoluciones de llamada se llenan explícitamente dentro del contexto, y cualquier devolución de llamada registrada se llama en el orden inverso cuando el flujo de control sale del contexto. El resultado es como tener múltipler declaraciones with anidadas, excepto que están establecidas dinamicamente.

Apilando gestores de contexto

Hay varias formas de llenar el ExitStack. Este ejemplo utiliza enter_context() para agregar un nuevo gestor de contexto a la pila.

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() primero llama __enter__() en el gestor de contexto, y luego registra su método __exit__() como una devolución de llamada para ser invocado cuando la pila se deshace.

$ python3 contextlib_exitstack_enter_context.py

0 entering
1 entering
inside context
1 exiting
0 exiting

Los gestores de contexto dados a ExitStack son tratados como si están en una serie de declaraciones with anidadas. Errores que ocurren en cualquier lugar dentro del contexto se propagan a través del manejo de error normal de los gestores de contexto. Estas clases de gestores de contexto ilustran la forma en que se propagan los errores.

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

Los ejemplos que usan estas clases se basan en variable_stack(), que usa los gestores de contexto pasados para construir un ExitStack, construyendo el contexto general uno por uno. Los ejemplos a continuación pasan diferentes gestores de contexto para explorar el comportamiento de manejo de errores. Primero, el caso normal sin excepciones.

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

Luego, un ejemplo de manejo de excepciones dentro de los gestores de contexto al final de la pila, en el que todos los contextos abiertos se cierran cuando la pila se desenrolla

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

A continuación, un ejemplo de manejo de excepciones dentro de los gestores de contexto en el medio de la pila, en el cual el error no ocurre hasta que los contextos ya están cerrados, por lo que esos contextos no ven el error.

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

Finalmente, un ejemplo de la excepción que queda sin ser manejada y se propaga hasta el código de llamada.

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

Si algún gestor de contexto en la pila recibe una excepción y regresa un valor True, evita que la excepción se propague a cualquier otro gestor de contexto.

$ 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

Devoluciones de llamada de contextos arbitrarios

ExitStack también admite devoluciones de llamada arbitrarias para cerrar un contexto, lo que facilita la limpieza de los recursos que no están controlados a través de un gestor de contexto.

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

Al igual que con los métodos __exit__() de los gestores de contexto completo, las devoluciones de llamada se invocan en el orden inverso en que son registradas.

$ python3 contextlib_exitstack_callbacks.py

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

Las devoluciones de llamada se invocan independientemente de si se produjo un error, y no se les da ninguna información sobre si ocurrió un error. Su valor de retorno es ignorado.

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

Debido a que no tienen acceso al error, las devoluciones de llamada no pueden suprimir excepciones de ser propagadas a través del resto de la pila de gestores de contexto.

$ python3 contextlib_exitstack_callbacks_error.py

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

Las devoluciones de llamadas son una forma conveniente de definir claramente la lógica de limpieza sin la sobrecarga de crear una nueva clase de gestor de contexto. Para mejorar la legibilidad del código, esa lógica puede ser encapsulada en función en línea, y se puede usar callback() como decorador.

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

No hay forma de especificar los argumentos para las funciones registradas usando la forma de decorador de callback(). Sin embargo, si la devolución de llamada de limpieza se la define en línea, las reglas de alcance le dan acceso a las variables definidas en el código de llamada.

$ python3 contextlib_exitstack_callbacks_decorator.py

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

Pilas parciales

A veces, al construir contextos complejos, es útil poder abortar una operación si el contexto no se puede construir por completo, y retrasar la limpieza de todos los recursos hasta un momento posterior si todos pueden configurarse correctamente. Por ejemplo, si una operación necesita varias conexiones de red duraderas, puede ser mejor no iniciar la operación si falla una conexión. Sin embargo, si todas las conexiones se pueden abrir necesitan permanecer abiertas más tiempo que la duración de un gestor de contexto único. El método pop_all() de ExitStack se puede usar en este escenario.

pop_all() borra todos los gestores de contexto y devoluciones de llamada de la pila en la que se llama, y devuelve una nueva pila pre-poblada con esos mismos gestores de contexto y devoluciones de llamada. El método close() de la nueva pila se puede invocar más tarde, después de que la pila original se ha ido, para limpiar los recursos.

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

Este ejemplo usa las mismas clases de gestor de contexto definidas anteriormente, con la diferencia de que ErrorOnEnter produce un error en __enter__() en lugar de en __exit__(). Dentro de variable_stack(), si se ingresa a todos los contextos sin error, entonces el método close() de un nuevo ExitStack es devuelto. Si ocurre un error manejado, variable_stack() regresa None para indicar que el trabajo de limpieza ya fue hecho. Y si se produce un error no controlado, la pila parcial se limpia y el error se propaga

$ 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

Ver también