Context Managers

What is a Context Manager?

A context manager manages resources (files, connections, locks) ensuring proper setup and teardown.

# Without context manager
file = open('data.txt', 'w')
try:
    file.write('Hello')
finally:
    file.close()  # Must explicitly close

# With context manager
with open('data.txt', 'w') as file:
    file.write('Hello')
# Automatically closed, even if exception occurs

The with Statement

with expression as variable:
    # Use the resource
    pass
# Resource is automatically cleaned up

# Multiple context managers
with open('input.txt') as infile, open('output.txt', 'w') as outfile:
    outfile.write(infile.read())

# Or using contextlib
from contextlib import ExitStack
with ExitStack() as stack:
    files = [stack.enter_context(open(f)) for f in filenames]

Why Use Context Managers?

1. Guaranteed cleanup: Resources are always released
2. Exception safety: Cleanup happens even on errors
3. Cleaner code: No try/finally boilerplate
4. Resource management: Database connections, locks, threads

enter and exit Methods

class FileManager:
    def __init__(self, filename, mode):
        self.filename = filename
        self.mode = mode
        self.file = None
    
    def __enter__(self):
        # Setup: open file, acquire lock, etc.
        self.file = open(self.filename, self.mode)
        return self.file
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        # Teardown: close file, release lock, etc.
        if self.file:
            self.file.close()
        # Return False to propagate exceptions
        # Return True to suppress exceptions
        return False

# Usage
with FileManager('test.txt', 'w') as f:
    f.write('Hello, World!')
# File is automatically closed

Exception Handling in Context Managers

class DatabaseConnection:
    def __init__(self, connection_string):
        self.connection_string = connection_string
        self.conn = None
    
    def __enter__(self):
        self.conn = create_connection(self.connection_string)
        return self.conn
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is not None:
            # Exception occurred - rollback
            self.conn.rollback()
            print(f'Error: {exc_val}')
        else:
            # No exception - commit
            self.conn.commit()
        
        self.conn.close()
        return False  # Don't suppress exception

# Usage
with DatabaseConnection('localhost') as conn:
    conn.execute('INSERT INTO users ...')
    # If exception: rollback and close
    # If success: commit and close

Reusable Context Manager

class Timer:
    def __init__(self, label=''):
        self.label = label
    
    def __enter__(self):
        import time
        self.start = time.perf_counter()
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        import time
        self.elapsed = time.perf_counter() - self.start
        if self.label:
            print(f'{self.label}: {self.elapsed:.4f}s')
        return False

# Reusable!
with Timer('First'):
    sum(range(1000000))

with Timer('Second'):
    sum(range(1000000))

@contextmanager Decorator

from contextlib import contextmanager

@contextmanager
def managed_resource(name):
    print(f'Acquiring {name}')
    resource = {'name': name, 'active': True}
    try:
        yield resource  # This is the 'as' value
    except Exception as e:
        print(f'Error: {e}')
        resource['active'] = False
    finally:
        print(f'Releasing {name}')

# Usage
with managed_resource('database') as res:
    print(f'Using {res["name"]}')
# Acquiring database
# Using database
# Releasing database

Common Patterns

from contextlib import contextmanager
import os

# Temporary directory
@contextmanager
def temporary_directory():
    import tempfile
    temp_dir = tempfile.mkdtemp()
    try:
        yield temp_dir
    finally:
        import shutil
        shutil.rmtree(temp_dir)

# Changed directory
@contextmanager
def changed_directory(path):
    old_dir = os.getcwd()
    os.chdir(path)
    try:
        yield
    finally:
        os.chdir(old_dir)

# Suppress exceptions
from contextlib import suppress

with suppress(FileNotFoundError):
    os.remove('nonexistent.txt')  # No error

Nested Context Managers

from contextlib import ExitStack

# Handle multiple resources
def process_files(filenames):
    with ExitStack() as stack:
        files = [stack.enter_context(open(f)) for f in filenames]
        # All files are open
        for f in files:
            print(f.read())
        # All files closed automatically

# Or using nested()
from contextlib import nested

# Python 3 only allows multiple with items
with open('a.txt') as a, open('b.txt') as b:
    pass

Closing Pattern

from contextlib import closing

class DatabaseCursor:
    def __init__(self, connection):
        self.connection = connection
    
    def execute(self, query):
        return self.connection.execute(query)
    
    def close(self):
        self.connection.close()

# Using closing()
with closing(DatabaseCursor(conn)) as cursor:
    cursor.execute('SELECT * FROM users')
# cursor.close() is automatically called