Context manager

1. Overview

Context managers allow you to allocate and release resources precisely when you want to. The most widely used example of context managers is the with a statement. Suppose you have two related operations which you’d like to execute as a pair, with a block of code in between. Context managers allow you to do specifically that. For example:

2. Description

2.1 Context

In python, context is the state surrounding a section of code. 

f = open('test.txt', 'r')
print(f.readlines())
f.close()

when print(f.readlines()) runs, it has a context in which it runs global scope.

2.2 Context manager

Create a context and execute some code that uses variables from the context. After that automatically clean up the context when we are done with it.

Context managers manage data in our scope of entry and exit. It is very useful for anything that needs to provide Enter/Exit, Start/Stop, and Set/Reset. For example, open/close file, start DB transaction/commit or abort a transaction, set decimal precision to 3 / reset back to original precision.

2.2.1 The context management protocol

Classes implement the context management protocol by implementing two methods

__enter__  # setup, and optionally return some object
__exit__   # tear down / cleanup

2.2.2 Use cases

Very common usage is for opening files (creating resource) and closing the file (releasing resource). Context managers can be used for much more than creating and releasing resources

2.2.3 Common patterns

• Open/Close
• Lock/Release
• Change/Reset
• Start/Stop
• Enter/Exit

2.2.4 Examples

• File context managers
• Decimal contexts

3. Statements

3.1 try and finally statement

The final section of a try always executes even if a function and return are in the try or except blocks. It is very useful for writing code that should execute no matter what happens. But this can get cumbersome.

3.2 with statement

In with statement, the object returned from context and after the with block, context is cleaned up automatically.

class MyClass:
    def __init__(self):
        #init class

    def __enter__(self):
        return obj

    def __exit__(self, + ...):
        #clean up obj
        
        
 
# works as a regular class __enter__, __exit__ were not called
my_obj = MyClass() 



with MyClass() as obj:
# creates an instance of of Myclass 
# no associated symbol, but an instance exits
# my_instance
# calls my_instance.__enter__()
# return value from __enter__ is assigned to obj(not the instance of MyClass that was created)
# after the with block, or if an exception occurs inside the with block
# my_instance.__exit__ is called

3.2.1 Scope of with block

The with block is not like a function or a comprehension. The scope of anything in the with block(including the object returned from __enter__) is in the same scope as the with statement itself.

3.2.2 The __enter__ Method

This method should perform whatever setup it needs to. It can optionally return an object as returned_obj. 

def __enter__(self):
	# code

3.2.3 The __exit__ Method

Runs even if an exception occurs in with block. It knows about any exceptions that occurred and needs to tell Python to silence the exception or let it propagate.

with MyContext() as obj:
	raise ValueError
print('done')

Scenario 1: __exit__ receives an error, performs some cleanup and silences error. And print statement runs no exception is seen.

Scenario 2: __exit__ receives an error, performs some cleanup and let's error propagate. And print statement does not run. The ValueException is seen.

def __exit__(self, exc_type, exc_val, exc_tb):
        # do clean up work here
        return True # or False

Needs three arguments

• The exception type that occurred. If any, None otherwise
• The exception value that occurred. If any, None otherwise
• The traceback object if an exception occurred. If any, None otherwise

Return True or False

• True: silence any raised exception
• False: do not silence a raised exception

4. Example

4.1 Finally statement

def my_func():
    try:
        1/0
    except:
        return
    finally:
        print('finally running...')

my_func()

# even though return statement runned, finally is executed in python
#finally running...

4.2 with statement

with open('test.txt', 'w') as f:
    print('inside with: file closed?', f.closed)
    raise ValueError()
    
    
# inside with: file closed? False
# ---------------------------------------------------------------------------

# ValueError                                Traceback (most recent call last)

# <ipython-input-4-e9388ca8409f> in <module>
#       1 with open('test.txt', 'w') as f:
#       2     print('inside with: file closed?', f.closed)
# ----> 3     raise ValueError()
#       4 
#       5 

# ValueError: 
    
print('after with: file closed?', f.closed)
# after with: file closed? True

4.3 __enter__ and __exit__ method

class MyContext:
    def __init__(self):
        self.obj = None
        
    def __enter__(self):
        print('entering context...')
        self.obj = 'the Return Object'
        return self.obj

    def __exit__(self, exc_type, exc_value, exc_traceback):
        print('exiting context...')
        if exc_type:
            print(f'*** Error occurred: {exc_type}, {exc_value}')
        return False  # do not suppress exceptions
        
        
 with MyContext() as obj:
    raise ValueError
    
    
# entering context...
# exiting context...
# *** Error occurred: <class 'ValueError'>, 
# ---------------------------------------------------------------------------
# ValueError                                Traceback (most recent call last)
# <ipython-input-14-39a69b57f322> in <module>()
#       1 with MyContext() as obj:
# ----> 2     raise ValueError

# ValueError: 

class Resource:
    def __init__(self, name):
        self.name = name
        self.state = None
        
        
class ResourceManager:
    def __init__(self, name):
        self.name = name
        self.resource = None
        
    def __enter__(self):
        print('entering context')
        self.resource = Resource(self.name)
        self.resource.state = 'created'
        return self.resource
    
    def __exit__(self, exc_type, exc_value, exc_traceback):
        print('exiting context')
        self.resource.state = 'destroyed'
        if exc_type:
            print('error occurred')
        return False
        
        
with ResourceManager('spam') as res:
    print(f'{res.name} = {res.state}')
print(f'{res.name} = {res.state}')


# entering context
# spam = created
# exiting context
# spam = destroyed

6. Caveat 

6.1 Work with lazy iterator

6.1.1 return 

def read_data():
    with open('nyc_parking_tickets_extract.csv') as f:
        return csv.reader(f, delimiter=',', quotechar='"')

for row in read_data():
    print(row)
    
# ---------------------------------------------------------------------------

# FileNotFoundError                         Traceback (most recent call last)

# <ipython-input-3-3dcb2ffbcff9> in <module>
#       3         return csv.reader(f, delimiter=',', quotechar='"')
#       4 
# ----> 5 for row in read_data():
#       6     print(row)
#       7 

# <ipython-input-3-3dcb2ffbcff9> in read_data()
#       1 def read_data():
# ----> 2     with open('nyc_parking_tickets_extract.csv') as f:
#       3         return csv.reader(f, delimiter=',', quotechar='"')
#       4 
#       5 for row in read_data():

# FileNotFoundError: [Errno 2] No such file or directory: 'nyc_parking_tickets_extract.csv'

6.1.2 list which is eager

def read_data():
    with open('nyc_parking_tickets_extract.csv') as f:
        return list(csv.reader(f, delimiter=',', quotechar='"'))

for row in read_data():
    print(row)

6.1.3 better way using yield from

def read_data():
    with open('nyc_parking_tickets_extract.csv') as f:
        yield from csv.reader(f, delimiter=',', quotechar='"')

for row in read_data():
    print(row)

7. Generator and Context managers

class GenCtxManager:
    def __init__(self, gen_func, *args, **kwargs):
        self._gen = gen_func(*args, **kwargs)
        
    def __enter__(self):
        return next(self._gen)
    
    def __exit__(self, exc_type, exc_value, exc_tb):
        try:
            next(self._gen)
        except StopIteration:
            pass
        return False
        
        
def open_file(fname, mode):
    try:
        print('opening file...')
        f = open(fname, mode)
        yield f
    finally:
        print('closing file...')
        f.close()
        
        
with GenCtxManager(open_file, 'test.txt', 'w') as f:
    print('writing to file...')
    f.write('testing...')
    
# opening file...
# writing to file...
# closing file...
    
with open('test.txt') as f:
    print(next(f))

# testing...

7.1 Context Manager Decorator

def open_file(fname, mode='r'):
    print('opening file...')
    f = open(fname, mode)
    try:
        yield f
    finally:
        print('closing file...')
        f.close()
        
class GenContextManager:
    def __init__(self, gen):
        self.gen = gen
        
    def __enter__(self):
        return next(self.gen)
        
    def __exit__(self, exc_type, exc_value, exc_tb):
        print('calling next to perform cleanup in generator')
        try:
            next(self.gen)
        except StopIteration:
            pass
        return False
        
file_gen = open_file('test.txt', 'w')

with GenContextManager(file_gen) as f:
    f.writelines('Sir Spamalot')
    
# opening file...
# calling next to perform cleanup in generator
# closing file...

file_gen = open_file('test.txt')
with GenContextManager(file_gen) as f:
    print(f.readlines())
    
# opening file...
# ['Sir Spamalot']
# calling next to perform cleanup in generator
# closing file...

Applying Decorator

def context_manager_dec(gen_fn):
    def helper(*args, **kwargs):
        gen = gen_fn(*args, **kwargs)
        ctx = GenContextManager(gen)
        return ctx
    return helper
    
@context_manager_dec
def open_file(fname, mode='r'):
    print('opening file...')
    f = open(fname, mode)
    try:
        yield f
    finally:
        print('closing file...')
        f.close()    
        
with open_file('test.txt') as f:
    print(f.readlines())
    
# opening file...
# ['Sir Spamalot']
# calling next to perform cleanup in generator
# closing file...

7.2 contextlib contextmanager

from contextlib import contextmanager

@contextmanager
def open_file(fname, mode='r'):
    print('opening file...')
    f = open(fname, mode)
    try:
        yield f
    finally:
        print('closing file...')
        f.close() 
        
with open_file('test.txt') as f:
    print(f.readlines())
    
    
    
from time import perf_counter, sleep

@contextmanager
def timer():
    stats = dict()
    start = perf_counter()
    stats['start'] = start
    try:
        yield stats
    finally:
        end = perf_counter()
        stats['end'] = end
        stats['elapsed'] = end - start
        
with timer() as stats:
    sleep(1)
    
print(stats)
# {'start': 5.417897319468211e-07, 'end': 1.005228270913287, 'elapsed': 1.005227729123555}

7.3 Nested Context Manager

7.3.1 User-defined NestedContextManager

from contextlib import contextmanager

@contextmanager
def open_file(f_name):
    print(f'opening file {f_name}')
    f = open(f_name)
    try:
        yield f
    finally:
        print(f'closing file {f_name}')
        f.close()

class NestedContexts:
    def __init__(self):
        self._exits = []
        
    def __enter__(self):
        return self

    def enter_context(self, ctx):
        self._exits.append(ctx.__exit__)
        value = ctx.__enter__()
        return value
        
    def __exit__(self, exc_type, exc_value, exc_tb):
        for exit in self._exits[::-1]:
            exit(exc_type, exc_value, exc_tb)
        return False
        
f_names = 'file1.txt', 'file2.txt', 'file3.txt'

with NestedContexts() as stack:
    files = [stack.enter_context(open_file(f_name)) for f_name in f_names]
    
    while True:
        try:
            rows = [next(f).strip('\n') for f in files]
        except StopIteration:
            break
        else:
            row = ','.join(rows)
            print(row
            
# opening file file1.txt
# opening file file2.txt
# opening file file3.txt
# file1_line1,file2_line1,file3_line1
# file1_line2,file2_line2,file3_line2
# file1_line3,file2_line3,file3_line3
# closing file file3.txt
# closing file file2.txt
# closing file file1.txt

7.3.2 contextlib ExitStack

from contextlib import ExitStack

f_names = 'file1.txt', 'file2.txt', 'file3.txt'

with ExitStack() as stack:
    files = [stack.enter_context(open_file(f_name))
            for f_name in f_names]
    while True:
        try:
            rows = [next(f).strip('\n') for f in files]
        except StopIteration:
            break
        else:
            row = ','.join(rows)
            print(row)
            
# opening file file1.txt
# opening file file2.txt
# opening file file3.txt
# file1_line1,file2_line1,file3_line1
# file1_line2,file2_line2,file3_line2
# file1_line3,file2_line3,file3_line3
# closing file file3.txt
# closing file file2.txt
# closing file file1.txt

8. References

https://mathbyteacademy.com

https://en.wikibooks.org/wiki/Python_Programming/Context_Managers

https://book.pythontips.com/en/latest/context_managers.html