The contextlib module contains utilities for working with context managers and the with statement.
A context manager is responsible for a resource within a code block, possibly creating it when the block is entered and then cleaning it up after the block is exited. For example, files support the context manager API to make it easy to ensure they are closed after all reading or writing is done.
In [4]:
with open('tmp/pymotw.txt', 'wt') as f:
f.write('contents go here')
A context manager is enabled by the with statement, and the API involves two methods. The __enter__() method is run when execution flow enters the code block inside the with. It returns an object to be used within the context. When execution flow leaves the with block, the __exit__() method of the context manager is called to clean up any resources being used.
In [5]:
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__()')
In [6]:
with Context():
print('Doing work in the context')
The __enter__() method can return any object to be associated with a name specified in the as clause of the with statement. In this example, the Context returns an object that uses the open context.
In [7]:
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()
The value associated with the variable c is the object returned by __enter__(), which is not necessarily the Context instance created in the with statement.
The __exit__() method receives arguments containing details of any exception raised in the with block.
In [8]:
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')
If the context manager can handle the exception, __exit__() should return a true value to indicate that the exception does not need to be propagated. Returning false causes the exception to be re-raised after __exit__() returns.
The class ContextDecorator adds support to regular context manager classes to let them be used as function decorators as well as context managers.
In [9]:
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')
One difference with using the context manager as a decorator is that the value returned by enter() is not available inside the function being decorated, unlike when using with and as. Arguments passed to the decorated function are available in the usual way.
Creating context managers the traditional way, by writing a class with __enter__() and __exit__() methods, is not difficult. But sometimes writing everything out fully is extra overhead for a trivial bit of context. In those sorts of situations, use the contextmanager() decorator to convert a generator function into a context manager.
In [10]:
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')
The generator should initialize the context, yield exactly one time, then clean up the context. The value yielded, if any, is bound to the variable in the as clause of the with statement. Exceptions from within the with block are re-raised inside the generator, so they can be handled there.
The context manager returned by contextmanager() is derived from ContextDecorator, so it also works as a function decorator.
In [11]:
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'))
The file class supports the context manager API directly, but some other objects that represent open handles do not. The example given in the standard library documentation for contextlib is the object returned from urllib.urlopen(). There are other legacy classes that use a close() method but do not support the context manager API. To ensure that a handle is closed, use closing() to create a context manager for it.
In [13]:
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)
It is frequently useful to ignore exceptions raised by libraries, because the error indicates that the desired state has already been achieved, or it can otherwise be ignored. The most common way to ignore exceptions is with a try:except statement with only a pass statement in the except block.
In [14]:
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')
The try:except form can be replaced with contextlib.suppress() to more explicitly suppress a class of exceptions happening anywhere in the with block.
In [15]:
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')
Poorly designed library code may write directly to sys.stdout or sys.stderr, without providing arguments to configure different output destinations. The redirect_stdout() and redirect_stderr() context managers can be used to capture output from functions like this, for which the source cannot be changed to accept a new output argument.
In [16]:
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())
Most context managers operate on one object at a time, such as a single file or database handle. In these cases, the object is known in advance and the code using the context manager can be built around that one object. In other cases, a program may need to create an unknown number of objects in a context, while wanting all of them to be cleaned up when control flow exits the context. ExitStack was created to handle these more dynamic cases.
An ExitStack instance maintains a stack data structure of cleanup callbacks. The callbacks are populated explicitly within the context, and any registered callbacks are called in the reverse order when control flow exits the context. The result is like having multple nested with statements, except they are established dynamically.
There are several ways to populate the ExitStack. This example uses enter_context() to add a new context manager to the stack.
In [21]:
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')
The context managers given to ExitStack are treated as though they are in a series of nested with statements. Errors that happen anywhere within the context propagate through the normal error handling of the context managers. These context manager classes illustrate the way errors propagate.
In [18]:
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')
The examples using these classes are based around variable_stack(), which uses the context managers passed to construct an ExitStack, building up the overall context one by one. The examples below pass different context managers to explore the error handling behavior. First, the normal case of no exceptions.
In [23]:
print('No errors:')
variable_stack([
HandleError(1),
PassError(2),
],"test error")
In [ ]: