Exceptions
Context Managers
Automatic Cleanup
You open a file, read it, then must close it. Forgetting to close leaks resources.
Context managers with with automatically close resources when done - even if
exceptions occur. No more manual try-finally for cleanup.
Basic with statement
Let Python handle resource cleanup.
basic_with.py
# Basic with statement usage
def main():
# Simulated file handling
class FakeFile:
def __init__(self, name, mode):
self.name = name
self.mode = mode
self.closed = False
def __enter__(self):
print(f" Opening {self.name}")
return self
def __exit__(self, exc_type, exc_val, exc_tb):
print(f" Closing {self.name}")
self.closed = True
return False # Don't suppress exceptions
def read(self):
if self.closed:
raise ValueError("File is closed")
return f"Contents of {self.name}"
print("Basic with statement:\n")
# Using with statement
with FakeFile("data.txt", "r") as f:
content = f.read()
print(f" Read: {content}")
print(f" File closed: {f.closed}")
# Without with statement (manual)
print("\nManual resource management:")
f2 = FakeFile("manual.txt", "r")
f2.__enter__()
try:
content = f2.read()
print(f" Read: {content}")
finally:
f2.__exit__(None, None, None)
# Multiple context managers
print("\nMultiple resources:")
with FakeFile("file1.txt", "r") as f1, FakeFile("file2.txt", "r") as f2:
print(f" Reading {f1.name}")
print(f" Reading {f2.name}")
print(" Both files closed")
# Exception handling
print("\nWith exception:")
try:
with FakeFile("error.txt", "r") as f:
print(f" Inside with block")
raise ValueError("Simulated error")
except ValueError as e:
print(f" Caught: {e}")
print(f" File was still closed: {f.closed}")
# Timer context
import time
class Timer:
def __enter__(self):
self.start = time.time()
print(" Timer started")
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.elapsed = time.time() - self.start
print(f" Timer stopped: {self.elapsed:.4f}s")
return False
print("\nTiming code:")
with Timer():
# Simulate work
total = sum(range(1000))
print(f" Computed sum: {total}")
if __name__ == "__main__":
main()
with open(file) as f: automatically closes file when block exits.
with statement
Automatic resource management: `with resource as r:`. Cleanup guaranteed.
Class-based context manager
Create your own with enter and exit.
class_based.py
# Class-based context managers
def main():
# Simple resource manager
class Resource:
def __init__(self, name):
self.name = name
self.acquired = False
def __enter__(self):
print(f" Acquiring {self.name}")
self.acquired = True
return self
def __exit__(self, exc_type, exc_val, exc_tb):
print(f" Releasing {self.name}")
self.acquired = False
return False # Propagate exceptions
def use(self):
if not self.acquired:
raise RuntimeError("Resource not acquired")
print(f" Using {self.name}")
print("Resource management:\n")
with Resource("Database") as db:
db.use()
print(f" Acquired: {db.acquired}")
print(f" After with: {db.acquired}")
# Context manager with return value
class Connection:
def __init__(self, host, port):
self.host = host
self.port = port
self.connected = False
def __enter__(self):
print(f" Connecting to {self.host}:{self.port}")
self.connected = True
return self # Returns self for use in 'as' clause
def __exit__(self, exc_type, exc_val, exc_tb):
print(f" Disconnecting from {self.host}:{self.port}")
self.connected = False
return False
def send(self, data):
if not self.connected:
raise RuntimeError("Not connected")
print(f" Sent: {data}")
print("\nConnection:")
with Connection("localhost", 8080) as conn:
conn.send("Hello")
conn.send("World")
# Exception information in __exit__
class ErrorHandler:
def __enter__(self):
print(" Entering context")
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type is None:
print(" Exiting normally")
else:
print(f" Exception occurred: {exc_type.__name__}")
print(f" Message: {exc_val}")
return False # Don't suppress
print("\nError handling:")
with ErrorHandler():
print(" No error")
print()
try:
with ErrorHandler():
print(" About to raise error")
raise ValueError("Test error")
except ValueError:
print(" Exception propagated")
# Suppress exception by returning True
class Suppressor:
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type is ValueError:
print(f" Suppressing {exc_type.__name__}: {exc_val}")
return True # Suppress ValueError
return False # Propagate other exceptions
print("\nSuppressing exceptions:")
with Suppressor():
print(" Raising ValueError")
raise ValueError("This will be suppressed")
print(" Execution continued")
# Lock-like context manager
class Lock:
def __init__(self, name):
self.name = name
self.locked = False
def __enter__(self):
print(f" Acquiring lock: {self.name}")
self.locked = True
return self
def __exit__(self, exc_type, exc_val, exc_tb):
print(f" Releasing lock: {self.name}")
self.locked = False
return False
print("\nLocking:")
lock = Lock("data_lock")
with lock:
print(f" Lock held: {lock.locked}")
# Critical section
print(f" Lock released: {lock.locked}")
if __name__ == "__main__":
main()
__enter__ sets up, __exit__ cleans up. Works with with statement.
__enter__ __exit__
Protocol methods for context managers. Enter returns resource, exit cleans up.
Decorator-based context manager
Simpler syntax with @contextmanager.
contextmanager_decorator.py
# contextmanager decorator
from contextlib import contextmanager
def main():
# Simple context manager with decorator
@contextmanager
def managed_resource(name):
print(f" Setup: {name}")
yield name # Provide value to 'as' clause
print(f" Cleanup: {name}")
print("Decorator-based context manager:\n")
with managed_resource("Resource1") as res:
print(f" Using: {res}")
# Context manager with exception handling
@contextmanager
def safe_operation(name):
print(f" Starting: {name}")
try:
yield name
except Exception as e:
print(f" Error in {name}: {e}")
raise # Re-raise
finally:
print(f" Finishing: {name}")
print("\nWith exception handling:")
with safe_operation("Operation1"):
print(" Working...")
print()
try:
with safe_operation("Operation2"):
print(" Working...")
raise ValueError("Something went wrong")
except ValueError:
print(" Exception caught in main")
# Timer context manager
import time
@contextmanager
def timer(label):
start = time.time()
yield
elapsed = time.time() - start
print(f" {label}: {elapsed:.4f}s")
print("\nTiming:")
with timer("Computation"):
total = sum(range(100000))
print(f" Sum: {total}")
# Temporary directory simulation
@contextmanager
def temp_directory(name):
print(f" Creating temp dir: {name}")
dir_path = f"/tmp/{name}"
try:
yield dir_path
finally:
print(f" Removing temp dir: {name}")
print("\nTemporary directory:")
with temp_directory("work_dir") as path:
print(f" Using directory: {path}")
# Do work in temp directory
# Database transaction simulation
@contextmanager
def transaction(db_name):
print(f" BEGIN TRANSACTION on {db_name}")
try:
yield
print(f" COMMIT on {db_name}")
except Exception as e:
print(f" ROLLBACK on {db_name} ({e})")
raise
print("\nDatabase transactions:")
with transaction("users_db"):
print(" INSERT INTO users...")
print(" UPDATE users...")
print()
try:
with transaction("orders_db"):
print(" INSERT INTO orders...")
raise RuntimeError("Constraint violation")
except RuntimeError:
print(" Transaction rolled back")
# Changing context
@contextmanager
def working_directory(path):
original = "/current/dir"
print(f" Changing to: {path}")
try:
yield path
finally:
print(f" Restoring to: {original}")
print("\nWorking directory:")
with working_directory("/new/path") as path:
print(f" Working in: {path}")
if __name__ == "__main__":
main()
yield separates setup from cleanup. Code before yield runs on enter.
@contextmanager
`from contextlib import contextmanager`. Decorator creates context manager from generator.
Multiple contexts
Manage several resources at once.
multiple_contexts.py
# Multiple context managers
from contextlib import contextmanager
def main():
# Nested with statements (old style)
@contextmanager
def resource(name):
print(f" Open: {name}")
yield name
print(f" Close: {name}")
print("Nested contexts (old style):\n")
with resource("Resource1"):
with resource("Resource2"):
with resource("Resource3"):
print(" Using all resources")
# Multiple contexts in one with (Python 3.1+)
print("\nMultiple in one with:")
with resource("R1"), resource("R2"), resource("R3"):
print(" Using all resources")
# File copy simulation
@contextmanager
def fake_file(name, mode):
print(f" Opening {name} ({mode})")
yield f"<{name} handle>"
print(f" Closing {name}")
print("\nFile copy:")
with fake_file("input.txt", "r") as src, fake_file("output.txt", "w") as dst:
print(f" Reading from {src}")
print(f" Writing to {dst}")
# Database connection and cursor
@contextmanager
def connection(db_name):
print(f" Connect to {db_name}")
yield f"<{db_name} connection>"
print(f" Disconnect from {db_name}")
@contextmanager
def cursor(conn):
print(f" Create cursor on {conn}")
yield f"<cursor>"
print(f" Close cursor")
print("\nDatabase operations:")
with connection("users_db") as conn:
with cursor(conn) as cur:
print(f" Execute query with {cur}")
# Cleanup order matters
class OrderedResource:
def __init__(self, name, order):
self.name = name
self.order = order
def __enter__(self):
print(f" [{self.order}] Enter: {self.name}")
return self
def __exit__(self, exc_type, exc_val, exc_tb):
print(f" [{self.order}] Exit: {self.name}")
return False
print("\nCleanup order:")
with OrderedResource("First", 1), \
OrderedResource("Second", 2), \
OrderedResource("Third", 3):
print(" [X] All acquired")
# Exception during acquisition
@contextmanager
def failing_resource(name, should_fail=False):
print(f" Acquiring {name}")
if should_fail:
raise RuntimeError(f"{name} failed to acquire")
yield name
print(f" Releasing {name}")
print("\nException during acquisition:")
try:
with failing_resource("R1"), \
failing_resource("R2", should_fail=True), \
failing_resource("R3"):
print(" Won't reach here")
except RuntimeError as e:
print(f" Error: {e}")
print(" Note: R1 was released, R3 never acquired")
# Nested transactions
@contextmanager
def savepoint(name):
print(f" SAVEPOINT {name}")
try:
yield
print(f" RELEASE SAVEPOINT {name}")
except Exception as e:
print(f" ROLLBACK TO SAVEPOINT {name}")
raise
print("\nNested transactions:")
try:
with savepoint("sp1"):
print(" Operation 1")
with savepoint("sp2"):
print(" Operation 2")
raise ValueError("Error in sp2")
except ValueError:
print(" Rolled back to sp1")
if __name__ == "__main__":
main()
with open(a) as f1, open(b) as f2: - all cleaned up properly.
contextlib utilities
Helpful functions for common patterns.
contextlib_utilities.py
# contextlib utilities
from contextlib import contextmanager, suppress, redirect_stdout, closing
import io
def main():
# suppress - ignore specific exceptions
print("Using suppress:\n")
from contextlib import suppress
# Without suppress
try:
int("not a number")
except ValueError:
pass # Silently ignore
print(" Attempt 1: Failed silently")
# With suppress
with suppress(ValueError):
int("not a number")
print(" Attempt 2: Failed silently")
# Suppress multiple exception types
with suppress(ValueError, TypeError, KeyError):
d = {}
value = d["missing_key"]
print(" Attempt 3: Suppressed KeyError")
# redirect_stdout - capture print output
print("\nRedirecting stdout:")
output = io.StringIO()
with redirect_stdout(output):
print("This goes to StringIO")
print("Not to console")
captured = output.getvalue()
print(f" Captured: {repr(captured)}")
# closing - ensure object is closed
class Resource:
def __init__(self, name):
self.name = name
self.closed = False
def close(self):
print(f" Closing {self.name}")
self.closed = True
print("\nUsing closing:")
with closing(Resource("res1")) as res:
print(f" Using {res.name}")
print(f" Closed: {res.closed}")
# nullcontext - conditional context manager
from contextlib import nullcontext
print("\nConditional context:")
use_lock = False
@contextmanager
def lock():
print(" Acquiring lock")
yield
print(" Releasing lock")
context = lock() if use_lock else nullcontext()
with context:
print(" Critical section (no lock)")
use_lock = True
context = lock() if use_lock else nullcontext()
with context:
print(" Critical section (with lock)")
# ExitStack - dynamic context managers
from contextlib import ExitStack
print("\nExitStack:")
@contextmanager
def file_context(name):
print(f" Open {name}")
yield name
print(f" Close {name}")
files = ["file1.txt", "file2.txt", "file3.txt"]
with ExitStack() as stack:
handles = [stack.enter_context(file_context(f)) for f in files]
print(f" All files opened: {handles}")
print(" All files closed")
# ExitStack with callbacks
print("\nExitStack with callbacks:")
with ExitStack() as stack:
stack.callback(lambda: print(" Callback 1"))
stack.callback(lambda: print(" Callback 2"))
print(" In context")
# Practical: resource pool
print("\nResource pool:")
def acquire_resources(count):
stack = ExitStack()
resources = []
try:
for i in range(count):
res = Resource(f"res{i}")
stack.enter_context(closing(res))
resources.append(res)
return stack.pop_all(), resources
except:
stack.close()
raise
with ExitStack() as stack:
mgr, resources = acquire_resources(3)
stack.enter_context(mgr)
print(f" Using {len(resources)} resources")
if __name__ == "__main__":
main()
suppress(), redirect_stdout(), closing() - ready-made context managers.
Exercise: practical.py
Build a database connection context manager