Contextlib
- provides utilities for common tasks involving the
with statement.
What it provides?
- contextmanager
- asynccontextmanager
- closing
- aclosing
- nullcontext
- suppress
- redirect_stdout
- redirect_stderr
- chdir
- ContextDecorator
Implementing a simple Context Manager (using class)
-
A class needs to implement
__enter__and__exit__methods to be used as a context manager. -
__enter__method is called when thewithblock is entered. It doesn't take any arguments. __exit__method is called when thewithblock is exited. It takes three arguments:exc_type: exception typeexc_value: exception valuetraceback: traceback object
- If no exception occurs, all three arguments are
None. Else, they contain the exception details.
class MyCtx:
def __init__(self, num):
self.num = num
def __enter__(self):
print("enter block executing")
return self
def __exit__(self, exc_type, exc_val, exc_tb):
print("exit block executing")
if __name__ == "__main__":
with MyCtx(4) as c:
print(f"value of {c.num=}")
print("-"*40)
contextmanager decorator
- A simpler way to create a context manager is to use the
contextmanagerdecorator from thecontextlibmodule. - You don't need to implement a whole new class just to be able to use
with statementswith it. function decorated with @contextmanagershouldyieldthe resource that needs to be managed.- Use try, except and finally blocks.
exceptionraised inside thewith blockcan be caught in theexcept block.finally blockis used to clean up the resource, after thewith blockis exited.
from contextlib import contextmanager
@contextmanager
def my_ctx(*args, **kwargs):
try:
print('Entering context')
yield 'hello'
except Exception as e:
print(f'Caught exception: {e}')
finally:
print('Exiting context')
if __name__ == '__main__':
with my_ctx() as val:
print(f'Value: {val}')
print("-"*50)
AsyncContextManager
- similar to
contextmanager, but forasyncfunctions. - Used with
async withstatements. - To implement, use
__aenter__and__aexit__methods.
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 ...')
Closing
- closing is a helper function that simplifies the use of resources that need to be closed properly, such as files, sockets, or database connections.
- It ensures that the resource gets closed automatically when you're done with it, even if an error occurs.
Why Use closing?
- Some objects don't support the with statement directly (they don't have a
__enter__or__exit__method). - The closing function makes it possible to use these objects in a with block, ensuring proper cleanup.
- When the block ends, closing calls the resource's
close() methodautomatically
from contextlib import contextmanager, closing
class MyCtx:
def __init__(self, name):
self.name = name
print("myctx function called")
def close(self):
print("myctx close function called")
if __name__ == "__main__":
with closing(MyCtx("deependu")) as m:
print(m.name)
expected Output:
Aclosing
aclosingis the async version ofclosing.
NullContext
do-nothingplaceholder that can be used withwith statement.- Used when you don't need to do anything in the
__enter__and__exit__methods. - Mostly used for maintaining consistency in the code.
- Whatever a function returns, it should return support
withstatement.
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
- in the above code, we wanted to use
with block, so we usednullcontextto maintain consistency.
Suppress
- Return a context manager that suppresses any of the specified exceptions if they occur in the body of a with statement and then resumes execution with the first statement following the end of the with statement.
from contextlib import suppress
with suppress(FileNotFoundError):
os.remove('somefile.tmp')
## equivalent to
# try:
# os.remove('somefile.tmp')
# except FileNotFoundError:
# pass
- To pass multiple exceptions, use a tuple.
from contextlib import suppress
if __name__ == "__main__":
with suppress((ValueError, FileNotFoundError)):
raise ValueError()
raise FileNotFoundError()
Redirect_stdout
- Instead of printing to the console, you can redirect the output to a file or a string.
- To do this, we can modify the
sys.stdoutobject to point to a different file or string.
- But, this is not a good practice, as it can lead to issues with other parts of the code that rely on
sys.stdout, if we forget to reset it back to the original value. redirect_stdoutprovides a context manager that temporarily redirectssys.stdoutto a different file or stream.
from contextlib import redirect_stdout
with open('output.txt', 'w') as f:
with redirect_stdout(f):
print('This is redirected to a file')
print('This is printed to the console')
Redirect_stderr
- Similar to
redirect_stdout, but forsys.stderr.
chdir
- This is a simple wrapper around chdir(), it
changes the current working directory upon entering and restores the old one on exit.
ContextDecorator
- Used to
create custom decoratorsthat can beused as context managers. @contextmanageris implemented using the same method.- Class inheriting from
ContextDecoratorhave to implement__enter__and__exit__as normal.
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()
# output: Starting
# The bit in the middle
# Finishing
with mycontext():
print('The bit in the middle')
# output: Starting
# The bit in the middle
# Finishing
How are they implemented?
nullcontext
class MyNullcontext:
def __enter__(self):
return None
def __exit__(self, exc_type, exc_value, traceback):
pass