with Statement and Context Managers
The with statement is Python’s standard mechanism for acquiring and releasing resources, ensuring that resources are properly cleaned up even when an exception occurs. This article explains how it works and how to implement your own context managers.
with Statement Basics
# File I/O: f.close() is called automatically when the with block exits
with open("data.txt", "r", encoding="utf-8") as f:
content = f.read()
# Managing multiple resources at once (Python 3.10+ supports parenthesized multi-line form)
with (
open("source.txt", "r", encoding="utf-8") as src,
open("dest.txt", "w", encoding="utf-8") as dst,
):
dst.write(src.read())The Context Manager Protocol
Any object that implements __enter__ and __exit__ is a context manager:
__enter__(self): Called when thewithblock is entered; its return value is bound to theasvariable.__exit__(self, exc_type, exc_val, exc_tb): Called when thewithblock exits, regardless of whether an exception occurred. ReturningTruesuppresses the exception (it does not propagate); returningFalseorNonere-raises it.
class DBConnection:
def __init__(self, dsn: str):
self.dsn = dsn
self.conn = None
def __enter__(self):
print(f"Connecting to database: {self.dsn}")
self.conn = {"connected": True} # simulated connection
return self.conn # bound to the as variable
def __exit__(self, exc_type, exc_val, exc_tb):
print(f"Closing connection (exception type: {exc_type})")
self.conn = None
return False # do not suppress exceptions
with DBConnection("postgres://localhost/app") as conn:
print(f"Using connection: {conn}")
# raise RuntimeError("Query failed") # __exit__ is called even if an exception occurscontextlib.contextmanager — Generator-based Implementation
The contextlib.contextmanager decorator lets you implement a context manager with a single generator function, which is more concise than defining a class:
from contextlib import contextmanager
@contextmanager
def managed_resource(name: str):
print(f"Acquiring resource: {name}") # equivalent to __enter__
resource = {"name": name}
try:
yield resource # the yielded value is bound to the as variable
finally:
print(f"Releasing resource: {name}") # equivalent to __exit__
with managed_resource("database connection") as r:
print(f"Using: {r['name']}")Code before the yield corresponds to __enter__, and code after it corresponds to __exit__. Placing the yield inside a try/finally block guarantees that cleanup runs whether or not an exception occurs.
Timer Example
import time
from contextlib import contextmanager
@contextmanager
def timer(label: str = ""):
start = time.perf_counter()
try:
yield
finally:
elapsed = time.perf_counter() - start
print(f"{label} elapsed: {elapsed:.4f}s")
with timer("data processing"):
import time
time.sleep(0.1)
# data processing elapsed: 0.1001sOther contextlib Utilities
suppress — Silently Ignore Specific Exceptions
from contextlib import suppress
from pathlib import Path
# Equivalent to: try: ... except FileNotFoundError: pass
with suppress(FileNotFoundError):
Path("nonexistent.txt").unlink()
# Suppress multiple exception types
with suppress(KeyError, IndexError):
data = {}
print(data["missing"]) # KeyError is silently ignoredExitStack — Dynamically Manage Multiple Contexts
When you need to decide at runtime how many contexts to open, ExitStack is the right tool:
from contextlib import ExitStack
files = ["a.txt", "b.txt", "c.txt"]
with ExitStack() as stack:
handles = [
stack.enter_context(open(f, "w", encoding="utf-8"))
for f in files
]
for i, fh in enumerate(handles):
fh.write(f"Content {i}\n")
# When the with block exits, all files are closed automaticallynullcontext — Placeholder Context (Python 3.7+)
from contextlib import nullcontext
def process(f=None):
ctx = nullcontext(f) if f else open("default.txt", "r")
with ctx as file:
print(file.read())Practical Example: Atomic File Write
from contextlib import contextmanager
from pathlib import Path
@contextmanager
def atomic_write(path: Path, encoding: str = "utf-8"):
"""Write to a temporary file first, then atomically replace the target on success; delete the temp file on failure."""
tmp = path.with_suffix(path.suffix + ".tmp")
try:
with tmp.open("w", encoding=encoding) as f:
yield f
tmp.replace(path) # atomic replacement
except Exception:
if tmp.exists():
tmp.unlink()
raise
with atomic_write(Path("config.json")) as f:
import json
json.dump({"version": 2}, f, indent=2)