Real-world context manager examples and patterns
Real-world context managers solve recurring problems: ensuring files are closed, transactions are committed or rolled back, locks are released, temporary files are cleaned up, and operations are timed. This article shows practical, production-ready patterns you can adapt to your own code. Each example is battle-tested and handles edge cases: exceptions, partial failures, and cleanup errors.
Understanding these patterns prepares you to write your own context managers that are robust, exception-safe, and maintainable. Context managers are not an advanced feature — they are a foundational pattern that every working Python program relies on, from the standard library's open() function to web frameworks and database drivers.
Pattern 1: Safe File Operations
The built-in open() function is a context manager. Always use it with with to guarantee file closure.
# Example 1: Reading and processing a file safely
def process_large_file(filename):
"""Read a file line by line, safely closing on exception."""
with open(filename, 'r') as f:
for line_number, line in enumerate(f, start=1):
try:
# Process line
value = int(line.strip())
yield value * 2
except ValueError:
print(f"Warning: Line {line_number} is not a valid integer")
# Example 2: Writing with automatic flush and close
def write_data_safely(filename, data):
"""Write data to a file, with guaranteed flush and close."""
with open(filename, 'w') as f:
for item in data:
f.write(f"{item}\n")
# File is flushed and closed here, even if an exception occurred
print(f"Successfully wrote {len(data)} items to {filename}")
# Example 3: Multiple file operations
def copy_with_filtering(source, destination, condition):
"""Copy lines from source to destination, filtering lines."""
with open(source, 'r') as src, open(destination, 'w') as dst:
for line in src:
if condition(line):
dst.write(line)
print(f"Copied {destination}")
# Test
import tempfile
import os
with tempfile.TemporaryDirectory() as tmpdir:
source_file = os.path.join(tmpdir, "source.txt")
dest_file = os.path.join(tmpdir, "dest.txt")
# Create source file
with open(source_file, 'w') as f:
f.write("apple\n")
f.write("apricot\n")
f.write("banana\n")
f.write("avocado\n")
# Filter and copy
copy_with_filtering(source_file, dest_file, lambda line: line.startswith('a'))
# Verify
with open(dest_file, 'r') as f:
print(f"Destination contents: {f.read()}")
# Output:
# Copied /tmp/.../dest.txt
# Destination contents: apple
# apricot
# avocado
Pattern 2: Database Transactions
A transaction context manager ensures data consistency: commit on success, rollback on exception.
# Example 4: Database transaction context manager
from contextlib import contextmanager
import sqlite3
@contextmanager
def database_transaction(db_path):
"""Open a database connection and manage a transaction."""
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
try:
yield cursor
conn.commit()
print("Transaction committed")
except Exception as e:
conn.rollback()
print(f"Transaction rolled back due to {type(e).__name__}")
raise
finally:
conn.close()
# Using the transaction context manager
import tempfile
with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as tmp:
db_file = tmp.name
try:
# Create a table
with database_transaction(db_file) as cursor:
cursor.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, age INTEGER)")
cursor.execute("INSERT INTO users (name, age) VALUES (?, ?)", ("Alice", 30))
cursor.execute("INSERT INTO users (name, age) VALUES (?, ?)", ("Bob", 25))
# Query with transaction
with database_transaction(db_file) as cursor:
cursor.execute("SELECT * FROM users")
rows = cursor.fetchall()
for row in rows:
print(f"User: {row}")
# Transaction with rollback on error
print("\nAttempting invalid transaction:")
try:
with database_transaction(db_file) as cursor:
cursor.execute("INSERT INTO users (name, age) VALUES (?, ?)", ("Charlie", 35))
# Simulate an error
raise ValueError("Age validation failed")
except ValueError as e:
print(f"Error: {e}")
# Verify Charlie was not inserted
with database_transaction(db_file) as cursor:
cursor.execute("SELECT COUNT(*) FROM users")
count = cursor.fetchone()[0]
print(f"Total users after failed transaction: {count}")
finally:
import os
os.unlink(db_file)
# Output:
# Transaction committed
# User: (1, 'Alice', 30)
# User: (2, 'Bob', 25)
# Transaction committed
#
# Attempting invalid transaction:
# Transaction rolled back due to ValueError
# Error: Age validation failed
# Transaction committed
# Total users after failed transaction: 2
Pattern 3: Lock Management
Locks are context managers, ensuring they are released even on exceptions.
import threading
from contextlib import contextmanager
# Example 5: Custom lock context manager with timeout
class TimeoutLock:
def __init__(self, timeout=None):
self.lock = threading.Lock()
self.timeout = timeout
def __enter__(self):
acquired = self.lock.acquire(timeout=self.timeout)
if not acquired:
raise TimeoutError(f"Could not acquire lock within {self.timeout} seconds")
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.lock.release()
return False
# Using the lock
lock = TimeoutLock(timeout=1)
shared_resource = []
def worker(worker_id):
for i in range(3):
with lock:
shared_resource.append(f"worker_{worker_id}_item_{i}")
print(f"Worker {worker_id}: appended item {i}")
threads = [threading.Thread(target=worker, args=(i,)) for i in range(2)]
for t in threads:
t.start()
for t in threads:
t.join()
print(f"Resource has {len(shared_resource)} items")
# Output (order may vary):
# Worker 0: appended item 0
# Worker 1: appended item 0
# Worker 0: appended item 1
# Worker 1: appended item 1
# Worker 0: appended item 2
# Worker 1: appended item 2
# Resource has 6 items
Pattern 4: Temporary State Changes
A context manager can temporarily change state, then restore it.
from contextlib import contextmanager
import os
# Example 6: Temporary working directory change
@contextmanager
def temporary_directory(new_dir):
"""Temporarily change the working directory."""
original_dir = os.getcwd()
try:
os.chdir(new_dir)
print(f"Changed to {new_dir}")
yield new_dir
finally:
os.chdir(original_dir)
print(f"Restored to {original_dir}")
# Using it
import tempfile
with tempfile.TemporaryDirectory() as tmpdir:
with temporary_directory(tmpdir):
print(f"Current directory: {os.getcwd()}")
# Do work in tmpdir
print(f"Back to: {os.getcwd()}")
# Output:
# Changed to /tmp/tmpXXXX
# Current directory: /tmp/tmpXXXX
# Restored to /original/directory
# Back to: /original/directory
Pattern 5: Timing and Performance Profiling
A context manager that measures execution time is incredibly useful.
from contextlib import contextmanager
import time
# Example 7: Simple timer context manager
@contextmanager
def timer(operation_name):
"""Measure the time taken by a code block."""
start = time.time()
print(f"{operation_name}: starting")
try:
yield
finally:
elapsed = time.time() - start
print(f"{operation_name}: completed in {elapsed:.3f}s")
# Using the timer
with timer("Sorting large list"):
# Simulate some work
data = list(range(10000000, 0, -1))
data.sort()
# Output:
# Sorting large list: starting
# Sorting large list: completed in 1.234s
Pattern 6: Suppress Expected Exceptions
A context manager can handle expected exceptions gracefully.
from contextlib import contextmanager
# Example 8: Suppress specific exceptions
@contextmanager
def suppress_and_log(*exception_types):
"""Suppress specified exceptions and log them."""
try:
yield
except exception_types as e:
print(f"Expected error (suppressed): {type(e).__name__}: {e}")
# Using it
print("--- With exception suppression ---")
with suppress_and_log(ZeroDivisionError, ValueError):
print("Attempting division by zero...")
result = 1 / 0 # Suppressed
print("Continued after exception")
print("\n--- Without suppression (different error) ---")
try:
with suppress_and_log(ZeroDivisionError):
result = int("not a number") # ValueError, not suppressed
except ValueError:
print("ValueError was not suppressed")
# Output:
# --- With exception suppression ---
# Attempting division by zero...
# Expected error (suppressed): ZeroDivisionError: division by zero
# Continued after exception
#
# --- Without suppression (different error) ---
# ValueError was not suppressed
Pattern 7: Atomic Operations
Ensure an operation is all-or-nothing: either fully complete or rollback.
from contextlib import contextmanager
import json
import tempfile
import os
# Example 9: Atomic file write
@contextmanager
def atomic_write(filepath):
"""Write to a temporary file, then atomically replace the original."""
temp_fd, temp_path = tempfile.mkstemp(dir=os.path.dirname(filepath))
try:
with os.fdopen(temp_fd, 'w') as temp_file:
yield temp_file
# If we reach here, the write succeeded
os.replace(temp_path, filepath)
print(f"Atomically wrote to {filepath}")
except Exception as e:
print(f"Write failed: {e}. Cleaning up temporary file.")
os.unlink(temp_path)
raise
# Using atomic write
with tempfile.TemporaryDirectory() as tmpdir:
filepath = os.path.join(tmpdir, "config.json")
# Successful write
with atomic_write(filepath) as f:
json.dump({"name": "App", "version": "1.0"}, f)
with open(filepath, 'r') as f:
print(f"File contents: {f.read()}")
# Failed write (temporary file cleaned up, original unchanged)
original_data = open(filepath, 'r').read()
print(f"\nOriginal data: {original_data}")
try:
with atomic_write(filepath) as f:
json.dump({"invalid": "data"}, f)
raise RuntimeError("Simulated error during write")
except RuntimeError:
print("Write operation failed")
# Verify original data is unchanged
with open(filepath, 'r') as f:
print(f"File still contains: {f.read()}")
# Output:
# Atomically wrote to /tmp/tmpXXXX/config.json
# File contents: {"name": "App", "version": "1.0"}
#
# Original data: {"name": "App", "version": "1.0"}
#
# Write failed: Simulated error during write. Cleaning up temporary file.
# Write operation failed
# File still contains: {"name": "App", "version": "1.0"}
Key Takeaways
- Always use
with open()for files to guarantee closure and exception safety. - Transaction context managers ensure commit on success, rollback on exception.
- Locks are context managers; always use
with lock:to prevent deadlocks from exceptions. - Temporary state changes (directory, stdout, settings) should be managed by context managers.
- Timing and profiling context managers provide insights into performance.
- Atomic writes and suppression patterns handle common edge cases elegantly.
Frequently Asked Questions
Should I ever use open() without with?
No, except in the REPL for quick testing. In production code, always use with. It prevents file handle leaks and makes exception safety explicit.
How do I handle partial failures in transactions?
Let the exception propagate and the rollback happens automatically. If you need custom rollback logic, implement it in the context manager's exception handler.
Can I nest locks safely?
Only if they are reentrant locks (threading.RLock). Regular locks will deadlock if acquired twice by the same thread. Use try-lock with timeout for potential deadlock scenarios.
How do I time async operations?
Use asyncio.get_event_loop().time() instead of time.time() in async context managers. Wrap it in @asynccontextmanager for async timing.
What is the difference between suppress and handling exceptions?
suppress (return True from __exit__) silently hides the exception. Handling logs it and re-raises. Always prefer logging and re-raising unless you have a very specific reason to suppress.