Context Managers in Python #

Context managers allow you to allocate and release resources precisely when you want to. The most common use is with the with statement.

The with Statement #

Automatically handle resource cleanup:

# Without context manager
file = open('example.txt', 'r')
try:
 content = file.read()
 print(content)
finally:
 file.close()
# With context manager
with open('example.txt', 'r') as file:
 content = file.read()
 print(content)
# File is automatically closed

Creating Context Managers with Classes #

Implement enter and exit methods:

class DatabaseConnection:
 def __init__(self, host, port):
 self.host = host
 self.port = port
 self.connection = None
 
 def __enter__(self):
 print(f"Connecting to {self.host}:{self.port}")
 self.connection = self._create_connection()
 return self.connection
 
 def __exit__(self, exc_type, exc_val, exc_tb):
 print("Closing connection")
 if self.connection:
 self.connection.close()
 
 # Return False to propagate exceptions
 # Return True to suppress exceptions
 return False
 
 def _create_connection(self):
 # Simulate connection creation
 return {"status": "connected"}
# Usage
with DatabaseConnection('localhost', 5432) as conn:
 print(f"Connection status: {conn['status']}")

Using contextlib #

Create context managers with the contextlib module:

from contextlib import contextmanager
@contextmanager
def file_manager(filename, mode):
 print(f"Opening {filename}")
 file = open(filename, mode)
 try:
 yield file
 finally:
 print(f"Closing {filename}")
 file.close()
# Usage
with file_manager('test.txt', 'w') as f:
 f.write('Hello, World!')
# Timer context manager
import time
@contextmanager
def timer(label):
 start = time.time()
 try:
 yield
 finally:
 end = time.time()
 print(f"{label}: {end - start:.2f} seconds")
with timer("Download"):
 # Simulate work
 time.sleep(2)

Multiple Context Managers #

Handle multiple resources in one statement:

# Multiple with statements (old way)
with open('input.txt', 'r') as infile:
 with open('output.txt', 'w') as outfile:
 outfile.write(infile.read())
# Multiple managers in one statement (Python 3+)
with open('input.txt', 'r') as infile, \
 open('output.txt', 'w') as outfile:
 outfile.write(infile.read())
# Using contextlib.ExitStack
from contextlib import ExitStack
def process_files(filenames):
 with ExitStack() as stack:
 files = [stack.enter_context(open(fname)) for fname in filenames]
 # All files are automatically closed when exiting
 for f in files:
 print(f.read())

Exception Handling #

Context managers can handle exceptions:

class ErrorHandler:
 def __enter__(self):
 return self
 
 def __exit__(self, exc_type, exc_val, exc_tb):
 if exc_type is None:
 print("No errors occurred")
 return False
 
 if exc_type == ValueError:
 print(f"Handled ValueError: {exc_val}")
 return True # Suppress the exception
 
 # Let other exceptions propagate
 return False
# Usage
with ErrorHandler():
 raise ValueError("This will be suppressed")
 print("This won't print")
print("Execution continues")

Async Context Managers #

Context managers for async code:

class AsyncDatabaseConnection:
 async def __aenter__(self):
 print("Connecting asynchronously")
 # await actual connection
 return self
 
 async def __aexit__(self, exc_type, exc_val, exc_tb):
 print("Disconnecting asynchronously")
 # await actual disconnection
 return False
# Usage with async/await
async def main():
 async with AsyncDatabaseConnection() as conn:
 # Do async work
 pass
# Using contextlib for async
from contextlib import asynccontextmanager
@asynccontextmanager
async def async_timer(label):
 import asyncio
 start = asyncio.get_event_loop().time()
 try:
 yield
 finally:
 end = asyncio.get_event_loop().time()
 print(f"{label}: {end - start:.2f} seconds")
async def example():
 async with async_timer("Operation"):
 await asyncio.sleep(1)

Common Use Cases #

Practical applications of context managers:

# 1. Lock management
from threading import Lock
lock = Lock()
with lock:
 # Critical section
 pass
# 2. Directory changes
import os
from contextlib import contextmanager
@contextmanager
def change_dir(path):
 old_dir = os.getcwd()
 os.chdir(path)
 try:
 yield
 finally:
 os.chdir(old_dir)
with change_dir('/tmp'):
 # Work in /tmp
 pass
# Back to original directory
# 3. Temporary environment variables
@contextmanager
def temp_env_var(key, value):
 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
# 4. Database transactions
@contextmanager
def transaction(connection):
 connection.begin()
 try:
 yield connection
 connection.commit()
 except Exception:
 connection.rollback()
 raise
with transaction(db_connection) as conn:
 conn.execute("INSERT INTO ...")