Skip to main content

Module 15 - Context Managers

Context managers provide a clean way to manage resources (files, network connections, database connections) by ensuring proper setup and cleanup. They guarantee that resources are properly released even if errors occur.


1. The with Statement

The with statement simplifies resource management by automatically handling setup and teardown operations.

Basic Syntax

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

File Handling with with

Without context manager (manual cleanup):

# Manual resource management
file = open('data.txt', 'r')
try:
content = file.read()
print(content)
finally:
file.close() # Must remember to close

With context manager (automatic cleanup):

# Automatic resource management
with open('data.txt', 'r') as file:
content = file.read()
print(content)
# File is automatically closed here
Benefit

The file is automatically closed when the with block exits, even if an exception occurs!


2. Common Built-in Context Managers

2.1 File Operations

# Reading files
with open('input.txt', 'r') as f:
data = f.read()

# Writing files
with open('output.txt', 'w') as f:
f.write('Hello, World!')

# Multiple files
with open('source.txt', 'r') as src, open('dest.txt', 'w') as dst:
dst.write(src.read())

2.2 Threading Locks

import threading

lock = threading.Lock()

with lock:
# Critical section - only one thread at a time
shared_resource += 1

2.3 Decimal Context

from decimal import Decimal, localcontext

with localcontext() as ctx:
ctx.prec = 2 # Set precision to 2 decimal places
result = Decimal('1.00') / Decimal('3.00')
print(result) # 0.33

3. Creating Custom Context Managers

3.1 Using Classes (__enter__ and __exit__)

To create a context manager, implement __enter__ and __exit__ methods.

class FileManager:
def __init__(self, filename, mode):
self.filename = filename
self.mode = mode
self.file = None

def __enter__(self):
"""Called when entering the 'with' block"""
print(f"Opening {self.filename}")
self.file = open(self.filename, self.mode)
return self.file

def __exit__(self, exc_type, exc_value, traceback):
"""Called when exiting the 'with' block"""
print(f"Closing {self.filename}")
if self.file:
self.file.close()
# Return False to propagate exceptions
return False

# Usage
with FileManager('test.txt', 'w') as f:
f.write('Hello from custom context manager!')

Output:

Opening test.txt
Closing test.txt

3.2 Database Connection Manager

import sqlite3

class DatabaseConnection:
def __init__(self, db_name):
self.db_name = db_name
self.conn = None

def __enter__(self):
self.conn = sqlite3.connect(self.db_name)
return self.conn.cursor()

def __exit__(self, exc_type, exc_value, traceback):
if exc_type is None:
# No exception - commit changes
self.conn.commit()
else:
# Exception occurred - rollback
self.conn.rollback()
self.conn.close()

# Usage
with DatabaseConnection('mydb.db') as cursor:
cursor.execute('CREATE TABLE IF NOT EXISTS users (id INTEGER, name TEXT)')
cursor.execute('INSERT INTO users VALUES (1, "Alice")')
# Auto-commit and close

3.3 Timer Context Manager

import time

class Timer:
def __enter__(self):
self.start = time.time()
return self

def __exit__(self, exc_type, exc_value, traceback):
self.end = time.time()
self.elapsed = self.end - self.start
print(f"Elapsed time: {self.elapsed:.4f} seconds")

# Usage
with Timer():
# Code to time
sum([i**2 for i in range(1000000)])

4. Context Managers with contextlib

The contextlib module provides utilities for creating context managers more easily.

4.1 Using @contextmanager Decorator

from contextlib import contextmanager

@contextmanager
def file_manager(filename, mode):
"""Simple file context manager using decorator"""
print(f"Opening {filename}")
file = open(filename, mode)
try:
yield file # This is what __enter__ returns
finally:
print(f"Closing {filename}")
file.close()

# Usage
with file_manager('test.txt', 'w') as f:
f.write('Using contextlib!')

4.2 Temporary Directory

import tempfile
import os

@contextmanager
def temporary_directory():
"""Create and clean up a temporary directory"""
temp_dir = tempfile.mkdtemp()
try:
yield temp_dir
finally:
# Clean up
import shutil
shutil.rmtree(temp_dir)

# Usage
with temporary_directory() as temp_dir:
print(f"Working in: {temp_dir}")
# Create files in temp_dir
with open(os.path.join(temp_dir, 'temp.txt'), 'w') as f:
f.write('Temporary data')
# temp_dir is automatically deleted

4.3 Suppressing Exceptions

from contextlib import suppress

# Ignore specific exceptions
with suppress(FileNotFoundError):
os.remove('non_existent_file.txt')
print("This runs even if file doesn't exist")

# Without suppress:
try:
os.remove('non_existent_file.txt')
except FileNotFoundError:
pass

4.4 Redirecting Output

from contextlib import redirect_stdout
import io

# Capture stdout
f = io.StringIO()
with redirect_stdout(f):
print("This goes to the StringIO object")
print("Not to the console")

output = f.getvalue()
print(f"Captured: {output}")

5. Advanced Context Manager Patterns

5.1 Nested Context Managers

# Traditional nesting
with open('input.txt', 'r') as input_file:
with open('output.txt', 'w') as output_file:
output_file.write(input_file.read())

# Using contextlib.ExitStack for dynamic nesting
from contextlib import ExitStack

with ExitStack() as stack:
files = [stack.enter_context(open(f'file{i}.txt', 'w')) for i in range(5)]
for i, f in enumerate(files):
f.write(f'Content {i}')
# All files automatically closed

5.2 Reentrant Context Manager

import threading

class ReentrantContextManager:
def __init__(self):
self.lock = threading.RLock() # Reentrant lock
self.count = 0

def __enter__(self):
self.lock.acquire()
self.count += 1
print(f"Acquired (count: {self.count})")
return self

def __exit__(self, exc_type, exc_value, traceback):
self.count -= 1
print(f"Released (count: {self.count})")
self.lock.release()

# Usage - can be nested
manager = ReentrantContextManager()
with manager:
print("Outer block")
with manager:
print("Inner block (reentrant)")

6. Comparison Table

ApproachProsConsUse Case
Class-basedFull control, reusableMore verboseComplex cleanup logic
@contextmanagerConcise, readableLess controlSimple scenarios
Built-in (open, etc.)Standard, optimizedLimited customizationCommon operations
contextlib utilitiesConvenient helpersMay obscure logicSpecific patterns

7. Common Use Cases

7.1 Change Directory Temporarily

import os

@contextmanager
def change_directory(path):
"""Temporarily change working directory"""
old_dir = os.getcwd()
try:
os.chdir(path)
yield
finally:
os.chdir(old_dir)

# Usage
print(f"Currently in: {os.getcwd()}")
with change_directory('/tmp'):
print(f"Now in: {os.getcwd()}")
print(f"Back to: {os.getcwd()}")

7.2 Environment Variable Management

@contextmanager
def env_variable(key, value):
"""Temporarily set an environment variable"""
old_value = os.environ.get(key)
os.environ[key] = value
try:
yield
finally:
if old_value is None:
del os.environ[key]
else:
os.environ[key] = old_value

# Usage
with env_variable('DEBUG', 'true'):
print(os.environ['DEBUG']) # 'true'

Summary

✅ Context managers ensure proper resource cleanup
with statement provides automatic setup and teardown
✅ Implement __enter__ and __exit__ for class-based managers
✅ Use @contextmanager decorator for simpler cases
contextlib provides useful utilities
✅ Always prefer context managers over manual cleanup


Next Steps

In Module 16, you'll learn:

  • Functional programming concepts
  • map, filter, and reduce
  • Lambda functions
  • Function composition

Practice Exercises

  1. Create a context manager that times code execution and logs the duration
  2. Build a DatabaseTransaction context manager that rolls back on errors
  3. Write a context manager that temporarily changes the logging level
  4. Create a context manager for managing multiple file operations atomically
  5. Implement a context manager that captures and suppresses specific warnings
Challenge

Create a ConnectionPool context manager that:

  • Maintains a pool of database connections
  • Provides a connection from the pool when entering
  • Returns the connection to the pool when exiting
  • Handles connection failures gracefully