Updated On : Feb-14,2021 Tags context-managers, with-statements
contextlib - Useful Context Managers (with Statement) in Python

contextlib - Useful Context Managers (with Statement) in Python

While coding in Python, we generally face situations where we need to use a particular object only for few lines of our code like using locks when updating shared data in a multithreading environment, using a file object to write some log to files, etc. Python supports this kind of object that defines a context in which a block of code related to that object will be executed. This concept of providing runtime context is generally referred to as context manager in python and is provided by using with statement. The context managers can be anything from shared resources, lock primitives, file pointer, etc. As a part of this tutorial, we'll start by explaining how we can create context managers in python and explore on Python module contextlib which provides a list of methods for working with context managers.

Python object can be referred to as context manager if it implements both of the below-mentioned methods.

  • __enter(self)__ - This method will be executed when context starts and returns the class instance itself that will be used as a context manager.
  • __exit(self, exc_type, exc_value, exc_tb)__ - This method will be executed when the context manager exits or code inside it fails and there is no exception handling. It has three parameters that will be set if there is an exception in code that was executed in this context else all will be set to None. It returns a boolean flag. It should return False if we want an error that happened in the code block inside the context to be further raised or return True if we have handled the error and don't want to propagate it further. If we return True then the statement after this context manager will start executing.
    • exc_type - exception type
    • exc_value - exception value
    • exc_tb - traceback

All the objects which implement these two methods can be considered context managers in Python and can be called using with statement.

Please make a NOTE that we'll be referring to these two methods as dunder (short of double underscore) enter and dunder exit going forward.

We'll now start explaining context manager with simple examples. We'll then go on to explain various methods and classes available in module contextlib with simple and easy to understand examples.

Example 1: Simple Context Manager Creation

As a part of our first example, we are demonstrating how we can create and use a simple context manager in Python.

We have created a simple class named SampleContextManager which implements enter and exit methods that we described earlier. It's a simple context manager that prints when context starts and ends. It'll print error details if there are any failures during the execution of the code block inside the context. It returns True hence making sure that line after it'll be executed. We can make changes to return False after printing the error if we want to further propagate the error.

We are then executing a simple print statement inside the context manager to see how it works. We can notice that it executes both enter and exit in order.

Then we are again creating context and executing statements that will raise an error. The error is handled by the exit dunder method which we can confirm from the output of the code.

In [8]:
class SampleContextManager:
    def __init__(self):
        pass

    def __enter__(self):
        print("Context Successfully Started\n")
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        if exc_type:
            print("Error Type : {}, Error : {}\n".format(exc_type.__name__, exc_value))
        else:
            print("Successfully Exited From Context\n")
        return True
In [9]:
with SampleContextManager() as sc:
    print("Statements Inside Context\n")
Context Successfully Started

Statements Inside Context

Successfully Exited From Context

In [10]:
with SampleContextManager() as sc:
    out = 10 / 0
Context Successfully Started

Error Type : ZeroDivisionError, Error : division by zero

Example 2: AbstractContextManager Example

Our second example is an exact copy of our first example with the only change that we are not inheriting from AbstractContextManager class of contextlib module. The AbstractContextManager is an abstract class that provides a default implementation of the dunder enter method which returns the class instance itself and the abstract dunder exit method which class extending it needs to implement.

Our code for this part when executed has the exactly same impact as the code from the previous example. If we extend AbstractContextManager, we can exclude the implementation of the dunder enter method if it’s not doing anything new then returning the class instance itself. We have kept it to show the sequence in which methods are executed.

If you are interested in learning about abstract base classes in Python in detail then please feel free to check our tutorial on the same.

In [11]:
import contextlib

class SampleContextManager(contextlib.AbstractContextManager):
    def __init__(self):
        pass

    def __enter__(self):
        print("Context Successfully Started\n")
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        if exc_type:
            print("Error Type : {}, Error : {}\n".format(exc_type.__name__, exc_value))
        else:
            print("Successfully Exited From Context\n")
        return True
In [12]:
with SampleContextManager() as sc:
    print("Statements Inside Context\n")
Context Successfully Started

Statements Inside Context

Successfully Exited From Context

In [13]:
with SampleContextManager() as sc:
    out = 10 / 0
Context Successfully Started

Error Type : ZeroDivisionError, Error : division by zero

Example 3: AbstractContextManager Example

Our third example again explains how to create a context manager by extending AbstractContextManager but this example is a little complicated to mimic a real-life scenario.

Queue Class

Our code has created a class named Queue which serves the purpose of the synchronized queue and takes two values for initiation. The list of values that will be initial queue members and a lock instance will be used to prevent threads from concurrently modifying queue data. This class will make sure that only one thread accesses the queue and modify its content at a time. We have introduced two new methods named push and pop which pushed an element from the queue and pops the first element from the queue. It’s a FIFO (First In, First Out) queue. It also has an implementation of dunder enter and exit methods like previous examples. We'll be using this class as a context manager.

push_nitems

This method takes as input a number, generates that many random numbers in the range 1-50 and push them to the queue. It sleeps for 1 second in between to mimic a real-life situation that takes time for the push operation.

Queue Context Manager

We are then creating a queue as context manager giving it an empty list as the initial queue and the lock that we have created using threading module. We have then created three threads that pushed 3,4 and 5 random items on the queue.

Please make a NOTE that this part of the code requires background on multithreading with Python. If you are interested in learning about how to work with threads in python then please feel free to check our tutorial on the same.

We can notice from that output that only one thread is able to access the queue at a time. We can notice other threads trying to access the lock as well when it’s already occupied.

If you just want to concentrate on contextlib and this example is blocking you then we would suggest to ignore this example and proceed further as it won't hurt your progress at all. You can come back and learn it later when you have enough background on threading.

In [47]:
import contextlib
import threading

class Queue(contextlib.AbstractContextManager):
    def __init__(self, queue, lock):
        self.lock = lock
        self.queue = queue if queue else []

    def __enter__(self):
        print("Context Successfully Started\n")
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        if exc_type:
            print("Thread {}. Error Type : {}, Error : {}\n".format(threading.current_thread().name,
                                                                    exc_type.__name__, exc_value))
        else:
            print("\nSuccessfully Exited From Context\n")
        return True

    def push(self, item):
        while self.lock.locked():
            print("Thread : {} tried to acquire lock but failed. Putting it to sleep.".format(threading.current_thread().name))
            time.sleep(2)

        self.lock.acquire()
        print("Push Op. Thread : {} acquired lock.".format(threading.current_thread().name))
        self.queue.append(item)
        print("Thread : {} pushed item {} to Queue.".format(threading.current_thread().name, item))
        self.lock.release()
        print("Push Op. Thread : {} released lock.".format(threading.current_thread().name))

    def pop(self):
        while self.lock.locked():
            print("Thread : {} tried to acquire lock but failed. Putting it to sleep.".format(threading.current_thread().name))
            time.sleep(2)

        self.lock.acquire()
        print("Pop Op. Thread : {} acquired lock.".format(threading.current_thread().name))
        item = self.queue.pop(0)
        print("Thread : {} popped item {} from Queue.".format(threading.current_thread().name, item))
        self.lock.release()
        print("Pop Op. Thread : {} released lock.".format(threading.current_thread().name))
        return item
In [48]:
import time
import random

def push_nitems(queue, n):
    for i in range(n):
        time.sleep(1)
        item = random.randint(1,50)
        queue.push(item)

lock = threading.Lock()

with Queue(queue=[], lock=lock) as queue:
    t1 = threading.Thread(target=push_nitems, args=(queue, 3), name="T1")
    t2 = threading.Thread(target=push_nitems, args=(queue, 4), name="T2")
    t3 = threading.Thread(target=push_nitems, args=(queue, 5), name="T3")

    t1.start()
    t2.start()
    t3.start()

    t1.join(), t2.join(), t3.join()

    print("\nQueue Length : {}, Queue : {}".format(len(queue.queue), queue.queue))
Context Successfully Started

Push Op. Thread : T1 acquired lock.
Thread : T1 pushed item 13 to Queue.
Push Op. Thread : T1 released lock.
Push Op. Thread : T2 acquired lock.
Thread : T2 pushed item 36 to Queue.
Push Op. Thread : T2 released lock.
Push Op. Thread : T3 acquired lock.
Thread : T3 pushed item 25 to Queue.
Push Op. Thread : T3 released lock.
Push Op. Thread : T2 acquired lock.
Thread : T2 pushed item 9 to Queue.
Push Op. Thread : T2 released lock.
Push Op. Thread : T3 acquired lock.
Thread : T1 tried to acquire lock but failed. Putting it to sleep.Thread : T3 pushed item 13 to Queue.
Push Op. Thread : T3 released lock.

Push Op. Thread : T2 acquired lock.
Thread : T2 pushed item 46 to Queue.
Push Op. Thread : T2 released lock.
Push Op. Thread : T3 acquired lock.
Thread : T3 pushed item 31 to Queue.
Push Op. Thread : T3 released lock.
Push Op. Thread : T2 acquired lock.
Thread : T2 pushed item 44 to Queue.
Push Op. Thread : T2 released lock.
Push Op. Thread : T3 acquired lock.Thread : T1 tried to acquire lock but failed. Putting it to sleep.

Thread : T3 pushed item 21 to Queue.
Push Op. Thread : T3 released lock.
Push Op. Thread : T3 acquired lock.
Thread : T3 pushed item 21 to Queue.
Push Op. Thread : T3 released lock.
Push Op. Thread : T1 acquired lock.
Thread : T1 pushed item 46 to Queue.
Push Op. Thread : T1 released lock.
Push Op. Thread : T1 acquired lock.
Thread : T1 pushed item 32 to Queue.
Push Op. Thread : T1 released lock.

Queue Length : 12, Queue : [13, 36, 25, 9, 13, 46, 31, 44, 21, 21, 46, 32]

Successfully Exited From Context

Example 4: contextmanager Decorator for Generating New Context Managers

As a part of our fourth example, we'll demonstrate how we can use contextmanager decorator of contextlib module to create a generic function that can be used to generate context managers. The function once decorated will be able to create a context manager and these context manager instance does not need to implement dunder enter and exit methods.

synch_primitive

It’s a mapping between synchronization primitive and class that implements it. We can create a primitive instance using this.

acquire_resource()

This method accepts resource name and argument to a class of that resource name. It then creates a resource instance, acquires it, and returns it. If the resource name provided is not present in mapping then it returns lock primitive by default.

release_resource()

This method accepts resource reference and releases it. It also handles any error if occurred during releasing it.

request_resource()

This method generates a context manager and yields it. It’s decorated with contextmanager. In order to treat resource instance returns by this method as a context manager, we need to yield it. This way it'll be treated as a context manager event though it does not have dunder enter and exit methods implemented. We can give any resource name to this method along with arguments, it'll acquire that resource and return it to us to be used as a context manager.

We are then trying to use different resources as a context manager. We are just printing the name of resource acquired in all examples which try different resources as a context manager.

In [51]:
import contextlib
import threading

synch_primitive = {
    "Lock": threading.Lock,
    "ReentrantLock": threading.RLock,
    "Condition": threading.Condition,
    "Semaphore": threading.Semaphore,
    "BoundedSemaphore": threading.BoundedSemaphore,

}

def acquire_resource(resource_type=None, **resource_args):
    resource_ref = synch_primitive.get(resource_type, None)
    if resource_type:
        print("Resource : {} requested. Arguments : {}".format(resource_type, resource_args))
        resource = resource_ref(**resource_args)
        resource.acquire()
    else:
        print("No resource requested. Allocating Lock Resource.")
        resource = threading.Lock()
        resource.acquire()
    return resource

def release_resource(resource):
    try:
        resource.release()
    except Exception as e:
        print("Error Type : {}, Error : {}".format(type(e).__name__, e))
    finally:
        print("Resource : {} released.".format(type(resource).__name__))

@contextlib.contextmanager
def request_resource(resource_type=None, **resource_args):
    try:
        resource = acquire_resource(resource_type, **resource_args)
        yield resource
    except Exception as e:
        print("Error Type : {}, Error : {}".format(type(e).__name__, e))
    finally:
        release_resource(resource)
In [52]:
with request_resource("Lock") as resource:
    print("Resource : {}".format(resource))
Resource : Lock requested. Arguments : {}
Resource : <locked _thread.lock object at 0x7f31d00eb468>
Resource : lock released.
In [103]:
with request_resource() as resource:
    print("Resource : {}".format(resource))
No resource requested. Allocating Lock Resource.
Resource : <locked _thread.lock object at 0x7fdddd5ea6c0>
Resource : lock released.
In [104]:
with request_resource("ReentrantLock") as resource:
    print("Resource : {}".format(resource))
Resource : ReentrantLock requested. Arguments : {}
Resource : <locked _thread.RLock object owner=140591384905536 count=1 at 0x7fdddd273a50>
Resource : RLock released.
In [105]:
with request_resource("Condition", lock=threading.Lock()) as resource:
    print("Resource : {}".format(resource))
Resource : Condition requested. Arguments : {'lock': <unlocked _thread.lock object at 0x7fdddd48ba58>}
Resource : <Condition(<locked _thread.lock object at 0x7fdddd48ba58>, 0)>
Resource : Condition released.
In [106]:
with request_resource("Semaphore", value=5) as resource:
    print("Resource : {}".format(resource))
Resource : Semaphore requested. Arguments : {'value': 5}
Resource : <threading.Semaphore object at 0x7fdddd21e2e8>
Resource : Semaphore released.
In [107]:
with request_resource("BoundedSemaphore", value=3) as resource:
    print("Resource : {}".format(resource))
Resource : BoundedSemaphore requested. Arguments : {'value': 3}
Resource : <threading.BoundedSemaphore object at 0x7fdddd227c88>
Resource : BoundedSemaphore released.

Example 5: Treat Instances with close() Method as Context Managers

As a part of our fifth example, we'll demonstrate how we can use closing() method to close objects who are not context managers but we want to treat them like one.

We have again created a SampleContextManager() which does not have a dunder enter and exit method implementation. Due to this, python does not allow us to be used as a context manager (with statement). But we want to use this class instance as context managers and it has a close method which we need to call when exiting from context.

We can use instances of class like this by wrapping them inside of closing method of contextlib module and they will be treated as a context manager. It'll call close() method of instance when exiting context. If won't wrap method inside the closing method and try to use as context manager then it'll raise an exception saying dunder enter method not found.

In [53]:
import contextlib

class SampleContextManager:
    def __init__(self):
        pass

    def close(self):
        print("Closing Resource")

with contextlib.closing(SampleContextManager()) as scm:
    print("Statements inside sontext manager goes here. Closing will call close() method on exit from context.")
Statements inside sontext manager goes here. Closing will call close() method on exit from context.
Closing Resource

Example 6: Optional Context Manager

As a part of our sixth example, we'll demonstrate how we can return the optional context manager if the original context manager is not available using nullcontext() method.

  • nullcontext(enter_result=None) - This method returns null context manager which returns parameter given to it when used as a context manager (with statement).

This method can be used in a situation where the original context manager was unavailable due to some reasons. It’s then up to the developer to handle resources that were requested through the context manager.

Below we have again used SampleContextManager from our previous examples. We have then created a method that requests context manager based on the resource name and processes list of values given to it within the context. We have put an if-else condition in the method to check if the requested resource is Lock or ReentrantLock, then return the original context manager else return the null context manager. Our code then uses returned context manager and processes a list of values given to the method by printing them in context.

We have called a method with different resource types to check which context manager it returns. We can see that in the case of the null context manager, it returns a string given to the nullcontext() as input. It would have returned None if we would not have given anything to the method.

In [7]:
import contextlib

class SampleContextManager:
    def __init__(self, resource_name):
        self.resource_name = resource_name

    def __enter__(self):
        print("Context Successfully Started\n")
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        if exc_type:
            print("Error Type : {}, Error : {}\n".format(exc_type.__name__, exc_value))
        else:
            print("Successfully Exited From Context\n")
        return True


def process_values(values, resource_name):
    if resource_name in ["Lock", "ReentrantLock"]:
        scm = SampleContextManager(resource_name)
    else:
        scm = contextlib.nullcontext(resource_name)

    print("Context Manager : {}\n".format(scm))
    with scm as manager:
        print("Resource : {}\n".format(manager))
        for value in values:
            print("Processing Value : {}".format(value))
        print()
In [8]:
process_values(["Val1", "Val2", "Val3"], "Lock")
Context Manager : <__main__.SampleContextManager object at 0x7fee04270128>

Context Successfully Started

Resource : <__main__.SampleContextManager object at 0x7fee04270128>

Processing Value : Val1
Processing Value : Val2
Processing Value : Val3

Successfully Exited From Context

In [9]:
process_values(["Val1", "Val2", "Val3"], "Semaphore")
Context Manager : <contextlib.nullcontext object at 0x7fee04270978>

Resource : Semaphore

Processing Value : Val1
Processing Value : Val2
Processing Value : Val3

Example 7: Suppress Errors using Context Managers

As a part of our seventh example, we'll demonstrate how we can use context managers to suppress errors inside of the context. The contextlib module provides us with method named suppress() for this purpose.

  • suppress() - This method accepts as many exception types that we want to suppress as input. It won't raise error and halt program execution, instead, it won’t execute lines after any failure in the code inside of context.

Below we have first used the method to suppress ZeroDivisionError error. We are executing statements inside the context which could raise that error.

We are then suppressing KeyError and ZeroDivisionError error. We have created a mapping with two key-value pairs. We are then trying to retrieve different values from the mapping some of which will fail with a key error. We have also included a method that raises ZeroDivisionError exception. We can notice from the output the context manager does not execute any statement after the first failure which happens when we try to retrieve the value of the key Semaphore from mapping. We have then the entry which could run successfully but it won't be executed.

In [15]:
import contextlib
import threading

with contextlib.suppress(ZeroDivisionError):
    out = 10/0

with contextlib.suppress(KeyError, ZeroDivisionError):
    mapping = {"Lock": threading.Lock(), "ReentrantLock": threading.RLock()}

    print("Multithreading Primitive : ", mapping["Lock"])

    print("Multithreading Primitive : ", mapping["Semaphore"])

    print("Multithreading Primitive : ", mapping["ReentrantLock"])

    print("Multithreading Primitive : ", mapping["Condition"])

    out = 10/0
Multithreading Primitive :  <unlocked _thread.lock object at 0x7fee0426ea30>

Example 8: Redirect Standard Output using Context Managers

We'll be using our eighth example to demonstrate how we can use context manager to direct our messages written to standard output to some other file or file-like object using redirect_stdout() method of contextlib.

  • redirect_stdout() - This method accepts a file-like object as input and redirects all messages written to standard output (sys.stdout) to the file pointed by it.

We have created a simple method named addition which takes two numbers as input, adds them, and prints the result to standard output. We have then created a context of the file and inside of it, we have created another context using redirect_stdout() method giving it a file object. We are then calling the addition method 3 times with different parameters inside of this context.

We can notice that output was not written to standard output and was redirected to file. This can be useful in situations when we temporarily want to redirect output to some other stream and can be easily done by using context manager.

In [36]:
import contextlib

def addition(a, b):
    print("Addition of {} & {} is {}".format(a,b, a+b))

with open("log.txt", "w") as f:
    with contextlib.redirect_stdout(f):
        addition(10,20)
        addition(20,20)
        addition(20,30)
In [37]:
!cat log.txt
Addition of 10 & 20 is 30
Addition of 20 & 20 is 40
Addition of 20 & 30 is 50

Example 9: Redirect Standard Errors using Context Managers

Our ninth example demonstrates how we can direct messages written to standard error stream to another file or file-like stream using redirect_stderr() method of contextlib.

  • redirect_stderr() - This method accepts a file-like object as input and redirects all messages written to standard output (sys.stderr) to the file pointed by it.

We have created a method named division which takes two parameters as input, divides them, and prints output to standard output. We have also done exception handling which writes an error message to the standard error stream in case of failures.

Please make a NOTE of file parameter of print() method which is set to sys.stderr in case of failure to direct messages to it. By default file parameter is set to sys.stdout in the method.

We have then created a context of the file and inside of its context of redirecting standard error messages to that file. We are calling the division method two times. One of them runs successfully and one will fail.

We can notice from the output that successfully is written to standard output whereas failure information is directed to file.

In [3]:
import contextlib
import sys

def division(a, b):
    try:
        print("Division of {} & {} is {}".format(a,b, a/b), file=sys.stdout)
    except Exception as e:
        print("Error Type : {}, Error : {}".format(type(e).__name__, e), file=sys.stderr)

with open("log.txt", "w") as f:
    with contextlib.redirect_stderr(f):
        division(10,20)
        division('10',20)
Division of 10 & 20 is 0.5
In [4]:
!cat log.txt
Error Type : TypeError, Error : unsupported operand type(s) for /: 'str' and 'int'

Example 10: ContextDecorator for Wrapping Function Code inside of Context

As a part of our tenth example, we'll explain how we can create context managers that can be used as function decorators so that all the code inside the function gets executed in that context. We need our context manager to simply extend ContextDecorator in order to be used as a decorator.

Our code for example creates SampleContextManager like previous examples but this time it extends ContextDecorator. We have then created a method named process_values which takes a list of values as input and prints them informing us that it’s processing them.

We have first called method without decorator and it just prints messages about the process. We have then decorated the method with our context manager and it executes those statements inside the context of the context manager which we can see based on messages printed. We have then also decorated the method again but this time we have an argument with the context manager to ask it to acquire resources. It performs the same way as the decorator without arguments but it was to explain that we can even use context managers with arguments as a decorator as well.

In [27]:
from contextlib import ContextDecorator

class SampleContextManager(ContextDecorator):
    def __init__(self, resource_name=None):
        self.resource_name = resource_name

    def __enter__(self):
        print("{} : Context Successfully Started\n".format(self.resource_name))
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        if exc_type:
            print("Error Type : {}, Error : {}\n".format(exc_type.__name__, exc_value))
        else:
            print("\n{} : Successfully Exited From Context\n".format(self.resource_name))
        return True
In [28]:
def process_values(values):
    for value in values:
        print("Processing Item : {}".format(value))
In [29]:
process_values(["T-shirts", "Pants", "Shirts"])
Processing Item : T-shirts
Processing Item : Pants
Processing Item : Shirts
In [30]:
@SampleContextManager()
def process_values(values):
    for value in values:
        print("Processing Item : {}".format(value))
In [31]:
process_values(["T-shirts", "Pants", "Shirts"])
None : Context Successfully Started

Processing Item : T-shirts
Processing Item : Pants
Processing Item : Shirts

None : Successfully Exited From Context

In [32]:
@SampleContextManager(resource_name="ReentrantLock")
def process_values(values):
    for value in values:
        print("Processing Item : {}".format(value))
In [33]:
process_values(["T-shirts", "Pants", "Shirts"])
ReentrantLock : Context Successfully Started

Processing Item : T-shirts
Processing Item : Pants
Processing Item : Shirts

ReentrantLock : Successfully Exited From Context

Example 11: ExitStack to Execute Code inside Bunch of Context Managers

As a part of our eleventh and last example, we'll explain how we can create a stack of context managers and execute them one by one using ExitStack(). This can be helpful when we have many context managers which can make code message if we keep on adding statements inside with statements. It'll add too many levels of indentations. The ExitStack() class provides an efficient way of handling nested context managers.

Below we have given important methods of ExitStack() class.

  • enter_context(cm) - It takes context manager instance and returns it after executing its dunder enter method.
  • push(cm) - It pushes context manager on the stack without executing its dunder enter method. Its dunder exit will be executed when exiting context though.
  • callback() - It lets us add function with arguments to be called when exiting from context.

Our code first simply creates a simple context manager that we have been using for the last many examples. We have then created a context manager using ExitStack(). Inside of that context manager we are simply looping through resources and entering their context by calling enter_context() method of exit stack instance giving it context manager instance. We are then just printing few print statements that we want to execute inside the combined context of all context which we started.

We can notice from the output that when we call enter_context() on each context manager, their dunder enter method gets called and returns context manager. Then all print statements of the context get executed. After statements of context as completed running, the dunder exit method of each context manager gets executed in REVERSE order. This is due to the stack of context managers. The same would happen if we would have used with context statements inside with statements.

The below code is exactly the same as the first part of our code:

with SampleContextManager("DB_Lock"):
    with SampleContextManager("Thread_Lock"):
        with SampleContextManager("Thread_Condition"):
            with SampleContextManager("File_Access"):
                print("="*80)
                print("List of statements which needs all these resources.")
                print("="*80)

Our second part of the code is almost the same as our first part but we have pushed one more context manager with the argument Total Cleanup using push() method. It'll make sure that its dunder exit method will be executed when we exit from context. We can notice that from that output and comparing it with the previous output.

Our third part of the code, simply adds a simple callback function using callback() method which prints resource names that will be released. The callback will be executed first before executing any dunder exit method of context managers. We can notice that from the output of the code.

In [6]:
import contextlib

class SampleContextManager:
    def __init__(self, resource_name):
        self.resource_name = resource_name

    def __enter__(self):
        print("{} : Context Successfully Started\n".format(self.resource_name))
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        if exc_type:
            print("Error Type : {}, Error : {}\n".format(exc_type.__name__, exc_value))
        else:
            print("{} : Successfully Exited From Context\n".format(self.resource_name))
        return True
In [5]:
with contextlib.ExitStack() as stack:
    for resource in ["DB_Lock", "Thread_Lock", "Thread_Condition", "File_Access"]:
        scm = stack.enter_context(SampleContextManager(resource))

    print("="*80)
    print("List of statements which needs all these resources.")
    print("="*80)
DB_Lock : Context Successfully Started

Thread_Lock : Context Successfully Started

Thread_Condition : Context Successfully Started

File_Access : Context Successfully Started

================================================================================
List of statements which needs all these resources.
================================================================================
File_Access : Successfully Exited From Context

Thread_Condition : Successfully Exited From Context

Thread_Lock : Successfully Exited From Context

DB_Lock : Successfully Exited From Context

In [39]:
with contextlib.ExitStack() as stack:
    for resource in ["DB_Lock", "Thread_Lock", "Thread_Condition", "File_Access"]:
        scm = stack.enter_context(SampleContextManager(resource))

    print("="*80)
    print("List of statements which needs all these resources.")
    print("="*80)

    stack.push(SampleContextManager("Total Cleanup"))
DB_Lock : Context Successfully Started

Thread_Lock : Context Successfully Started

Thread_Condition : Context Successfully Started

File_Access : Context Successfully Started

================================================================================
List of statements which needs all these resources.
================================================================================
Total Cleanup : Successfully Exited From Context

File_Access : Successfully Exited From Context

Thread_Condition : Successfully Exited From Context

Thread_Lock : Successfully Exited From Context

DB_Lock : Successfully Exited From Context

In [37]:
def callback(*args):
    print("This callback function will be executed when exiting from context managers.")
    print("Releasing Resources : {}\n".format(args))

with contextlib.ExitStack() as stack:
    resources = ["DB_Lock", "Thread_Lock", "Thread_Condition", "File_Access"]
    for resource in resources:
        scm = stack.enter_context(SampleContextManager(resource))

    #stack.push(SampleContextManager("Total_Cleanup"))
    stack.callback(callback, *resources)
    print("="*80)
    print("List of statements which needs all these resources.")
    print("="*80)
DB_Lock : Context Successfully Started

Thread_Lock : Context Successfully Started

Thread_Condition : Context Successfully Started

File_Access : Context Successfully Started

================================================================================
List of statements which needs all these resources.
================================================================================
This callback function will be executed when exiting from context managers.
Releasing Resources : ('DB_Lock', 'Thread_Lock', 'Thread_Condition', 'File_Access')

File_Access : Successfully Exited From Context

Thread_Condition : Successfully Exited From Context

Thread_Lock : Successfully Exited From Context

DB_Lock : Successfully Exited From Context

This ends our small tutorial explaining how we can create context managers to how we can use different methods of contextlib module to create different kinds of context managers. Please feel free to let us know your views in the comments section.

References



Sunny Solanki  Sunny Solanki