Python — Context Managers
A context manager in Python is an object that defines a temporary runtime context for a block of code. It handles setup and teardown automatically — acquiring a resource before the block and releasing it after, even if an exception occurs.
The with statement is the entry point. It eliminates the need for explicit try/finally blocks for common resource management patterns: file handles, database connections, locks, temporary directories, and more.
Python provides two ways to create context managers: class-based (implementing __enter__ and __exit__) and generator-based (using @contextmanager from contextlib). The standard library also ships utility context managers for common patterns.
The with statement simplifies resource management. The expression after with must produce a context manager; its __enter__ runs before the block and __exit__ runs after, regardless of how the block exits.
| 1 | # Basic syntax — no target variable |
| 2 | with open("file.txt", "w") as f: |
| 3 | f.write("hello") |
| 4 | |
| 5 | # Multiple context managers (Python 3.1+) |
| 6 | with open("a.txt") as f1, open("b.txt") as f2: |
| 7 | data = f1.read() + f2.read() |
| 8 | |
| 9 | # Parenthesised grouping (Python 3.10+) |
| 10 | with ( |
| 11 | open("src.txt") as src, |
| 12 | open("dst.txt", "w") as dst, |
| 13 | ): |
| 14 | dst.write(src.read()) |
| 15 | |
| 16 | # Equivalent try/finally (what with replaces) |
| 17 | f = open("file.txt", "w") |
| 18 | try: |
| 19 | f.write("hello") |
| 20 | finally: |
| 21 | f.close() |
The as clause binds the return value of __enter__ to a variable. This is typically the resource itself, but can be any object the manager chooses to expose.
Any class that implements __enter__ and __exit__ is a context manager. __enter__ receives no arguments beyond self and returns the value bound to as. __exit__ receives the exception type, value, and traceback — or three Nones if no exception occurred.
| 1 | class ManagedFile: |
| 2 | def __init__(self, path: str, mode: str = "r"): |
| 3 | self.path = path |
| 4 | self.mode = mode |
| 5 | self.file = None |
| 6 | |
| 7 | def __enter__(self): |
| 8 | self.file = open(self.path, self.mode) |
| 9 | return self.file # bound to "as" |
| 10 | |
| 11 | def __exit__(self, exc_type, exc_val, exc_tb): |
| 12 | if self.file: |
| 13 | self.file.close() |
| 14 | # Return False (or None) to propagate exceptions |
| 15 | # Return True to suppress them |
| 16 | return False |
| 17 | |
| 18 | with ManagedFile("hello.txt", "w") as f: |
| 19 | f.write("Hello, context manager!") |
__exit__ Return Value
| 1 | class SuppressValueError: |
| 2 | def __enter__(self): |
| 3 | return self |
| 4 | |
| 5 | def __exit__(self, exc_type, exc_val, exc_tb): |
| 6 | if exc_type is ValueError: |
| 7 | print(f"Suppressed: {exc_val}") |
| 8 | return True # <-- suppresses the exception |
| 9 | return False # re-raises other exceptions |
| 10 | |
| 11 | with SuppressValueError(): |
| 12 | raise ValueError("This will be suppressed") |
| 13 | |
| 14 | print("Execution continues here") |
Returning True from __exit__ tells Python to suppress the exception. This is rarely a good idea — it hides bugs. Only suppress when you have explicitly handled the exception inside __exit__.
The @contextmanager decorator lets you write a context manager as a generator function with a single yield. Code before the yield becomes __enter__; code after becomes __exit__.
| 1 | from contextlib import contextmanager |
| 2 | |
| 3 | @contextmanager |
| 4 | def managed_file(path: str, mode: str = "r"): |
| 5 | file = open(path, mode) |
| 6 | try: |
| 7 | yield file # value for "as" |
| 8 | finally: |
| 9 | file.close() |
| 10 | |
| 11 | with managed_file("hello.txt", "w") as f: |
| 12 | f.write("Hello from generator!") |
If an exception occurs inside the with block, it is re-raised at the yield statement inside the generator. Wrap the yield in try/finally or try/except/finally to handle cleanup or suppress exceptions.
| 1 | @contextmanager |
| 2 | def suppress(exception_type): |
| 3 | try: |
| 4 | yield |
| 5 | except exception_type: |
| 6 | pass # swallow the exception |
| 7 | |
| 8 | with suppress(ValueError): |
| 9 | raise ValueError("Gone") |
| 10 | |
| 11 | with suppress(FileNotFoundError): |
| 12 | open("nonexistent.txt") |
Generator-based managers are more concise but cannot be subclassed or reused as easily as class-based ones. Prefer them for simple, single-use wrappers.
The contextlib module provides ready-made context managers and helper tools for common patterns.
contextlib.suppress
| 1 | from contextlib import suppress |
| 2 | import os |
| 3 | |
| 4 | # Suppress specific exceptions |
| 5 | with suppress(FileNotFoundError): |
| 6 | os.remove("temp_file.txt") |
| 7 | |
| 8 | # Multiple exception types |
| 9 | with suppress(FileNotFoundError, PermissionError): |
| 10 | os.remove("protected_file.txt") |
| 11 | |
| 12 | # Equivalent verbose version: |
| 13 | try: |
| 14 | os.remove("temp_file.txt") |
| 15 | except FileNotFoundError: |
| 16 | pass |
contextlib.redirect_stdout / redirect_stderr
| 1 | from contextlib import redirect_stdout, redirect_stderr |
| 2 | import io |
| 3 | |
| 4 | buf = io.StringIO() |
| 5 | with redirect_stdout(buf): |
| 6 | print("This goes to buf") |
| 7 | print("Still to buf") |
| 8 | |
| 9 | output = buf.getvalue() # "This goes to buf\nStill to buf\n" |
| 10 | |
| 11 | # Redirect both stdout and stderr |
| 12 | err_buf = io.StringIO() |
| 13 | with redirect_stdout(buf), redirect_stderr(err_buf): |
| 14 | print("stdout message") |
| 15 | raise SystemExit("stderr message") # goes to err_buf |
contextlib.nullcontext
| 1 | from contextlib import nullcontext |
| 2 | |
| 3 | def process(data: str, use_compression: bool = False): |
| 4 | # Returns either a real context manager or a no-op |
| 5 | cm = gzip.open("data.gz", "wt") if use_compression else nullcontext() |
| 6 | with cm as f: |
| 7 | f.write(data) if use_compression else None |
| 8 | |
| 9 | # nullcontext also accepts a return value |
| 10 | with nullcontext("fallback") as val: |
| 11 | print(val) # "fallback" |
contextlib.ExitStack
| 1 | from contextlib import ExitStack |
| 2 | |
| 3 | # Dynamically manage an unknown number of context managers |
| 4 | def open_many(files: list[str]) -> list: |
| 5 | with ExitStack() as stack: |
| 6 | handles = [ |
| 7 | stack.enter_context(open(fname)) |
| 8 | for fname in files |
| 9 | ] |
| 10 | # All handles are open inside the with block |
| 11 | # They are closed in reverse order on exit |
| 12 | return handles # BAD — handles closed after with exit! |
| 13 | |
| 14 | # Correct pattern: keep the stack alive |
| 15 | def process_many(files: list[str], processor): |
| 16 | with ExitStack() as stack: |
| 17 | handles = [ |
| 18 | stack.enter_context(open(fname)) |
| 19 | for fname in files |
| 20 | ] |
| 21 | processor(handles) # safe — stack still alive |
| 22 | |
| 23 | # Callback-based cleanup |
| 24 | stack = ExitStack() |
| 25 | stack.callback(lambda: print("cleanup")) |
| 26 | stack.close() # runs all registered callbacks |
contextlib.closing
| 1 | from contextlib import closing |
| 2 | from urllib.request import urlopen |
| 3 | |
| 4 | # closing() calls obj.close() on exit |
| 5 | with closing(urlopen("https://example.com")) as resp: |
| 6 | data = resp.read() |
| 7 | |
| 8 | # Equivalent to: |
| 9 | resp = urlopen("https://example.com") |
| 10 | try: |
| 11 | data = resp.read() |
| 12 | finally: |
| 13 | resp.close() |
Use closing for objects that have a close() method but are not context managers themselves.
Nested with statements are common when multiple resources must be managed. Python supports both explicit nesting and comma-separated (Python 3.1+) or parenthesised (Python 3.10+) syntax.
| 1 | # Deep nesting (hard to read) |
| 2 | with open("a.txt") as a: |
| 3 | with open("b.txt") as b: |
| 4 | with open("c.txt") as c: |
| 5 | data = a.read() + b.read() + c.read() |
| 6 | |
| 7 | # Flat comma separation (Python 3.1+) |
| 8 | with open("a.txt") as a, open("b.txt") as b, open("c.txt") as c: |
| 9 | data = a.read() + b.read() + c.read() |
| 10 | |
| 11 | # Parenthesised grouping (Python 3.10+, preferred for >2) |
| 12 | with ( |
| 13 | open("a.txt") as a, |
| 14 | open("b.txt") as b, |
| 15 | open("c.txt") as c, |
| 16 | ): |
| 17 | data = a.read() + b.read() + c.read() |
| 18 | |
| 19 | # ExitStack for dynamic nesting |
| 20 | filenames = ["a.txt", "b.txt", "c.txt"] |
| 21 | with ExitStack() as stack: |
| 22 | files = [stack.enter_context(open(f)) for f in filenames] |
| 23 | data = "".join(f.read() for f in files) |
info
Order of Exit: Nested context managers exit in reverse order of entry — the same ordering as pushing and popping a stack. This is consistent whether you use nesting, commas, or parentheses.
Async context managers use __aenter__ and __aexit__ (both coroutines) with the async with statement. They are essential for managing asynchronous resources like database connections, HTTP sessions, and file handles in async code.
| 1 | import asyncio |
| 2 | from contextlib import asynccontextmanager |
| 3 | |
| 4 | # Class-based async context manager |
| 5 | class AsyncDatabase: |
| 6 | async def __aenter__(self): |
| 7 | self.conn = await create_connection() |
| 8 | return self.conn |
| 9 | |
| 10 | async def __aexit__(self, exc_type, exc_val, exc_tb): |
| 11 | await self.conn.close() |
| 12 | |
| 13 | async def query(): |
| 14 | async with AsyncDatabase() as db: |
| 15 | return await db.execute("SELECT 1") |
| 16 | |
| 17 | # Generator-based (@asynccontextmanager) |
| 18 | @asynccontextmanager |
| 19 | async def async_managed_file(path: str, mode: str = "r"): |
| 20 | f = await open_async(path, mode) |
| 21 | try: |
| 22 | yield f |
| 23 | finally: |
| 24 | await f.close() |
| 25 | |
| 26 | async def read_data(): |
| 27 | async with async_managed_file("data.txt") as f: |
| 28 | return await f.read() |
| 29 | |
| 30 | # Async ExitStack (AsyncExitStack) |
| 31 | from contextlib import AsyncExitStack |
| 32 | |
| 33 | async def manage_many(): |
| 34 | async with AsyncExitStack() as stack: |
| 35 | db = await stack.enter_async_context(AsyncDatabase()) |
| 36 | f = await stack.enter_async_context(async_managed_file("log.txt")) |
| 37 | # both managed until the async with block exits |
info
async with — aenter / aexit: In Python 3.5+, __aenter__ should return awaitable and __aexit__ should accept (self, exc_type, exc_val, exc_tb) and return awaitable. The @asynccontextmanager decorator (Python 3.7+) mirrors @contextmanager for async generators.
File Handling
| 1 | from contextlib import contextmanager |
| 2 | import tempfile, shutil, os |
| 3 | |
| 4 | @contextmanager |
| 5 | def temp_dir(): |
| 6 | """Create a temporary directory and clean it up.""" |
| 7 | path = tempfile.mkdtemp() |
| 8 | try: |
| 9 | yield path |
| 10 | finally: |
| 11 | shutil.rmtree(path) |
| 12 | |
| 13 | with temp_dir() as workdir: |
| 14 | path = os.path.join(workdir, "data.txt") |
| 15 | with open(path, "w") as f: |
| 16 | f.write("temporary data") |
| 17 | # directory + contents removed after this block |
Database Connection
| 1 | from contextlib import contextmanager |
| 2 | |
| 3 | @contextmanager |
| 4 | def db_session(connection_string: str): |
| 5 | """Manage a DB session with automatic rollback on error.""" |
| 6 | conn = create_engine(connection_string).connect() |
| 7 | transaction = conn.begin() |
| 8 | try: |
| 9 | yield conn |
| 10 | transaction.commit() |
| 11 | except: |
| 12 | transaction.rollback() |
| 13 | raise |
| 14 | finally: |
| 15 | conn.close() |
| 16 | |
| 17 | with db_session("postgresql://localhost/mydb") as conn: |
| 18 | conn.execute("INSERT INTO users (name) VALUES ('Alice')") |
| 19 | # committed on success, rolled back on exception |
Threading Lock
| 1 | import threading |
| 2 | |
| 3 | counter = 0 |
| 4 | lock = threading.Lock() |
| 5 | |
| 6 | def safe_increment(): |
| 7 | global counter |
| 8 | with lock: # acquires on enter, releases on exit |
| 9 | counter += 1 |
| 10 | |
| 11 | # threading.Lock is itself a context manager |
| 12 | # Equivalent to: |
| 13 | # lock.acquire() |
| 14 | # try: |
| 15 | # counter += 1 |
| 16 | # finally: |
| 17 | # lock.release() |
| 18 | |
| 19 | # RLock (re-entrant lock) also works as a context manager |
| 20 | rlock = threading.RLock() |
| 21 | with rlock: |
| 22 | with rlock: # same thread can re-acquire |
| 23 | print("Nested lock is safe with RLock") |
Timing / Profiling
| 1 | import time |
| 2 | from contextlib import contextmanager |
| 3 | |
| 4 | @contextmanager |
| 5 | def timed(label: str = "block"): |
| 6 | start = time.perf_counter() |
| 7 | try: |
| 8 | yield |
| 9 | finally: |
| 10 | elapsed = time.perf_counter() - start |
| 11 | print(f"${label}: ${elapsed:.3f}s") |
| 12 | |
| 13 | with timed("sleeper"): |
| 14 | time.sleep(0.5) |
| 15 | # Output: sleeper: 0.500s |
| 16 | |
| 17 | # Measure multiple blocks |
| 18 | with timed("query"): |
| 19 | db.execute("SELECT * FROM big_table") |
| 20 | with timed("transform"): |
| 21 | transform(results) |
Context managers are a powerful abstraction, but they come with gotchas and guidelines worth following.
Do's
- Use with for every resource that supports it — files, locks, connections, subprocesses, and network sockets.
- Prefer @contextmanager for simple, single-purpose managers; use classes when you need inheritance, state, or composition.
- Keep the with block as narrow as possible to release resources promptly.
- Use ExitStack when the number of context managers is dynamic or unknown at write time.
- Use suppress instead of bare try/except: pass — it makes intent explicit.
- Use nullcontext to conditionally enable resource management without code duplication.
Don'ts
- Do not return True from __exit__ unless you are absolutely certain the exception should be suppressed — it hides errors from callers.
- Do not yield more than once in a @contextmanager generator — the second yield raises RuntimeError.
- Do not keep references to the resource after the with block exits — the resource is already released.
- Do not use contextmanager on a generator that does not yield — it will raise RuntimeError.
warning
Resource Lifetimes: A common bug is returning a managed resource from a with block. The resource is closed on exit, so the caller receives a dead object. If you need to extend the lifetime, manage it at the caller level or use ExitStack to pop individual managers.
| 1 | # BAD: resource closed before caller can use it |
| 2 | def open_bad(path: str): |
| 3 | with open(path) as f: |
| 4 | return f # f is closed here! |
| 5 | |
| 6 | # GOOD: caller manages the context |
| 7 | def open_good(path: str): |
| 8 | return open(path) |
| 9 | |
| 10 | with open_good("data.txt") as f: # caller owns the lifecycle |
| 11 | data = f.read() |