Python Context Managers Practice Problems & Exercises
Practice: Context Managers
← Back to lessonEasy
Use a with statement to write "Hello, context managers!" to a file, read it back, and verify the file is closed after the block exits.
with open("greeting.txt", "w") as f:
f.write("Hello, context managers!")
with open("greeting.txt", "r") as f:
content = f.read()
print(content)
print(f"File closed: {f.closed}")Solution
with open("greeting.txt", "w") as f:
f.write("Hello, context managers!")
with open("greeting.txt", "r") as f:
content = f.read()
print(content)
print(f"File closed: {f.closed}")
Output:
Hello, context managers!
File closed: True
How it works: The with statement calls f.__enter__() when entering the block, which returns the file object bound to f. When the block exits (normally or via exception), it calls f.__exit__(), which closes the file. After the block, f still exists as a variable, but f.closed is True — the resource has been released.
Key insight: Without with, you must remember to call f.close() manually, and if an exception occurs between open() and close(), the file leaks. The with statement guarantees cleanup regardless of how the block exits.
# Write "Hello, context managers!" to a file called "greeting.txt"
# using a with statement, then read it back and print the content.
# After the with block, print whether the file is closed.
Expected Output
Hello, context managers!
File closed: TrueHints
Hint 1: The with statement automatically calls close() on the file when the block exits.
Hint 2: open() returns a file object that supports the context manager protocol.
Hint 3: After the with block exits, the file handle is closed — verify with f.closed.
Use a single with statement with two context managers to open a source file for reading and a destination file for writing simultaneously. Copy all lines and print the count.
with open("source.txt", "w") as f:
f.write("line 1\n")
f.write("line 2\n")
f.write("line 3\n")
with open("source.txt", "r") as src, open("dest.txt", "w") as dst:
lines = src.readlines()
dst.writelines(lines)
print(f"Copied {len(lines)} lines")Solution
with open("source.txt", "w") as f:
f.write("line 1\n")
f.write("line 2\n")
f.write("line 3\n")
with open("source.txt", "r") as src, open("dest.txt", "w") as dst:
lines = src.readlines()
dst.writelines(lines)
print(f"Copied {len(lines)} lines")
Output:
Copied 3 lines
How it works: The single with statement manages two file handles. Both src and dst are opened when entering the block, and both are closed when exiting. The cleanup happens in reverse order — dst closes first, then src. This is equivalent to nesting two with statements but is more concise.
Key insight: Under the hood, with A as a, B as b: is equivalent to with A as a: with B as b:. If opening B raises an exception, A is still properly closed because its __exit__ runs during stack unwinding.
# First, create a source file with 3 lines
with open("source.txt", "w") as f:
f.write("line 1\n")
f.write("line 2\n")
f.write("line 3\n")
# Now use a SINGLE with statement to open source.txt for reading
# AND dest.txt for writing, then copy all lines from source to dest.
# Print how many lines were copied.
Expected Output
Copied 3 linesHints
Hint 1: Python 3 allows multiple context managers in a single with statement, separated by commas.
Hint 2: Both files are guaranteed to close when the with block exits.
Hint 3: Use readlines() to get a list of lines, then writelines() to write them all.
Implement a Timer class that acts as a context manager. It should measure the elapsed time inside the with block and store it in self.elapsed. Print whether the elapsed time is greater than zero.
import time
class Timer:
def __enter__(self):
self.start = time.perf_counter()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.end = time.perf_counter()
self.elapsed = self.end - self.start
return False
with Timer() as t:
total = sum(range(1000000))
print(f"Elapsed: {t.elapsed > 0}")Solution
import time
class Timer:
def __enter__(self):
self.start = time.perf_counter()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.end = time.perf_counter()
self.elapsed = self.end - self.start
return False
with Timer() as t:
total = sum(range(1000000))
print(f"Elapsed: {t.elapsed > 0}")
Output:
Elapsed: True
How it works: __enter__ records the start time using time.perf_counter() (the highest-resolution clock available) and returns self so the as t binding gives access to the Timer instance. __exit__ records the end time and computes the difference. Returning False (or None) means exceptions are not suppressed — they propagate normally.
Key insight: The __exit__ method always receives three arguments: exc_type, exc_val, and exc_tb. If no exception occurred, all three are None. If an exception occurred, they contain the exception details. Returning True suppresses the exception; returning False (or any falsy value) lets it propagate.
import time
class Timer:
# Implement __enter__ and __exit__
# __enter__: record start time, return self
# __exit__: record end time, compute self.elapsed
pass
# Use the Timer context manager
# After the block, print whether elapsed is greater than 0
Expected Output
Elapsed: TrueHints
Hint 1: __enter__ should record the start time and return self.
Hint 2: __exit__ should record the end time and compute the elapsed duration.
Hint 3: Use time.perf_counter() for high-resolution timing.
Write a change_dir(path) context manager using @contextmanager that temporarily changes the working directory. It should yield the previous directory path and restore it on exit.
import os
from contextlib import contextmanager
@contextmanager
def change_dir(path):
old_dir = os.getcwd()
os.chdir(path)
try:
yield old_dir
finally:
os.chdir(old_dir)
original = os.getcwd()
with change_dir("/tmp") as old:
inside = os.getcwd()
print(f"Changed: {inside == '/tmp'}")
print(f"Restored: {os.getcwd() == original}")Solution
import os
from contextlib import contextmanager
@contextmanager
def change_dir(path):
old_dir = os.getcwd()
os.chdir(path)
try:
yield old_dir
finally:
os.chdir(old_dir)
original = os.getcwd()
with change_dir("/tmp") as old:
inside = os.getcwd()
print(f"Changed: {inside == '/tmp'}")
print(f"Restored: {os.getcwd() == original}")
Output:
Changed: True
Restored: True
How it works: The @contextmanager decorator transforms a generator function into a context manager. Everything before yield is the setup (__enter__), and the yielded value becomes what as old binds to. Everything after yield is the teardown (__exit__). The try/finally ensures the directory is restored even if an exception occurs inside the with block.
Key insight: Without try/finally, if the code inside the with block raises an exception, the os.chdir(old_dir) after yield would never execute, leaving the process in the wrong directory. Always use try/finally in @contextmanager generators when cleanup is critical.
import os
from contextlib import contextmanager
# Implement a change_dir(path) generator-based context manager
# Before yield: save current dir with os.getcwd(), change with os.chdir(path)
# yield the old directory path
# After yield: change back to the saved directory (use try/finally!)
# Test: create a temp directory, use the context manager, verify restoration
Expected Output
Changed: True
Restored: TrueHints
Hint 1: Import contextmanager from contextlib.
Hint 2: The code before yield runs on enter — save the old directory and change to the new one.
Hint 3: The code after yield runs on exit — change back to the saved directory.
Hint 4: Use a try/finally to guarantee the directory is restored even if an exception occurs.
Medium
Implement a SelectiveSuppressor context manager that suppresses only specified exception types. When a suppressed exception occurs, print a message. When a non-suppressed exception occurs, print a different message and let it propagate.
class SelectiveSuppressor:
def __init__(self, *exception_types):
self.exception_types = exception_types
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type is None:
return False
if issubclass(exc_type, self.exception_types):
print(f"Caught {exc_type.__name__}: {exc_val}")
return True
print(f"Propagating {exc_type.__name__}: {exc_val}")
return False
with SelectiveSuppressor(ValueError, KeyError):
raise ValueError("bad value")
print("After ValueError block - program continues")
try:
with SelectiveSuppressor(ValueError, KeyError):
raise TypeError("wrong type")
except TypeError as e:
print(f"Caught propagated TypeError: {e}")Solution
class SelectiveSuppressor:
def __init__(self, *exception_types):
self.exception_types = exception_types
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type is None:
return False
if issubclass(exc_type, self.exception_types):
print(f"Caught {exc_type.__name__}: {exc_val}")
return True
print(f"Propagating {exc_type.__name__}: {exc_val}")
return False
with SelectiveSuppressor(ValueError, KeyError):
raise ValueError("bad value")
print("After ValueError block - program continues")
try:
with SelectiveSuppressor(ValueError, KeyError):
raise TypeError("wrong type")
except TypeError as e:
print(f"Caught propagated TypeError: {e}")
Output:
Caught ValueError: bad value
After ValueError block - program continues
Propagating TypeError: wrong type
Caught propagated TypeError: wrong type
How it works: __exit__ checks whether the raised exception is a subclass of any type in self.exception_types. If yes, it prints a message and returns True, which tells Python to suppress the exception — execution continues after the with block. If no, it returns False, letting the exception propagate up the call stack.
Key insight: This is essentially what contextlib.suppress(*exceptions) does in the standard library. The issubclass check means that suppressing ValueError also suppresses its subclasses like UnicodeError. This mirrors how except ValueError catches subclasses too.
class SelectiveSuppressor:
# Implement __enter__ and __exit__
# __init__ takes a tuple of exception types to suppress
# __exit__: if the exception type is in the suppress list,
# print a message and return True (suppress)
# otherwise print a different message and return False (propagate)
pass
# Test 1: suppress ValueError
# Test 2: do NOT suppress TypeError — catch it outside
Expected Output
Caught ValueError: bad value
After ValueError block - program continues
Propagating TypeError: wrong type
Caught propagated TypeError: wrong typeHints
Hint 1: __exit__ receives exc_type, exc_val, exc_tb when an exception occurs inside the with block.
Hint 2: Return True from __exit__ to suppress the exception — execution continues after the with block.
Hint 3: Return False to let the exception propagate normally.
Hint 4: You can selectively suppress based on the exception type.
Write a transaction(db) context manager using @contextmanager that commits on success and rolls back on exception. The exception should be re-raised after rollback.
from contextlib import contextmanager
class FakeDB:
def commit(self):
print("COMMIT")
def rollback(self, reason):
print(f"ROLLBACK due to: {reason}")
@contextmanager
def transaction(db):
print("Starting transaction")
try:
yield db
db.commit()
except Exception as e:
db.rollback(str(e))
raise
db = FakeDB()
with transaction(db) as conn:
print("Inserting user: Alice")
result = "Alice inserted"
print(f"Transaction result: {result}")
try:
with transaction(db) as conn:
print("Inserting user: Error")
raise RuntimeError("Simulated DB error")
except RuntimeError as e:
print(f"Caught: {e}")Solution
from contextlib import contextmanager
class FakeDB:
def commit(self):
print("COMMIT")
def rollback(self, reason):
print(f"ROLLBACK due to: {reason}")
@contextmanager
def transaction(db):
print("Starting transaction")
try:
yield db
db.commit()
except Exception as e:
db.rollback(str(e))
raise
db = FakeDB()
with transaction(db) as conn:
print("Inserting user: Alice")
result = "Alice inserted"
print(f"Transaction result: {result}")
try:
with transaction(db) as conn:
print("Inserting user: Error")
raise RuntimeError("Simulated DB error")
except RuntimeError as e:
print(f"Caught: {e}")
Output:
Starting transaction
Inserting user: Alice
COMMIT
Transaction result: Alice inserted
Starting transaction
Inserting user: Error
ROLLBACK due to: Simulated DB error
Caught: Simulated DB error
How it works: The @contextmanager generator pauses at yield db, handing control to the with block. If the block completes normally, execution resumes after yield and db.commit() runs. If the block raises an exception, the except clause catches it, calls db.rollback(), and re-raises with raise. This is the commit/rollback pattern used in SQLAlchemy, Django, and most database frameworks.
Key insight: The db.commit() line sits right after yield — this means it only executes if yield completes without an exception (i.e., the with block succeeded). This is a subtle but powerful placement: the happy path is the code right after yield, and the error path is in the except block.
from contextlib import contextmanager
class FakeDB:
def commit(self):
print("COMMIT")
def rollback(self, reason):
print(f"ROLLBACK due to: {reason}")
# Implement a transaction(db) context manager using @contextmanager
# Before yield: print "Starting transaction", yield the db
# On success (no exception): call db.commit()
# On failure: call db.rollback(str(exception)), then re-raise
# Test 1: successful transaction
# Test 2: failed transaction (catch the re-raised exception)
Expected Output
Starting transaction
Inserting user: Alice
COMMIT
Transaction result: Alice inserted
Starting transaction
Inserting user: Error
ROLLBACK due to: Simulated DB error
Caught: Simulated DB errorHints
Hint 1: Use @contextmanager and yield the connection/transaction object.
Hint 2: In the try block: yield, then commit on success.
Hint 3: In the except block: rollback on failure, then re-raise the exception.
Hint 4: The finally block can be used for shared cleanup.
Implement an IndentLogger class that is both reusable and reentrant. Each nesting level adds indentation. The same instance can be used in multiple and nested with statements.
class IndentLogger:
def __init__(self):
self.level = 0
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.level -= 1
return False
def log(self, message):
indent = " " * self.level
print(f"{indent}[{self.level}] {message}")
def indent(self):
self.level += 1
return self
logger = IndentLogger()
print("--- Reusable (IndentLogger) ---")
logger.log("Starting")
with logger.indent():
logger.log("Level 1")
with logger.indent():
logger.log("Level 2")
logger.log("Back to 1")
logger.log("Done")
print("--- Reuse same instance ---")
logger.log("Fresh start")
with logger.indent():
logger.log("Nested again")
logger.log("End")Solution
class IndentLogger:
def __init__(self):
self.level = 0
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.level -= 1
return False
def log(self, message):
indent = " " * self.level
print(f"{indent}[{self.level}] {message}")
def indent(self):
self.level += 1
return self
logger = IndentLogger()
print("--- Reusable (IndentLogger) ---")
logger.log("Starting")
with logger.indent():
logger.log("Level 1")
with logger.indent():
logger.log("Level 2")
logger.log("Back to 1")
logger.log("Done")
print("--- Reuse same instance ---")
logger.log("Fresh start")
with logger.indent():
logger.log("Nested again")
logger.log("End")
Output:
--- Reusable (IndentLogger) ---
[0] Starting
[1] Level 1
[2] Level 2
[1] Back to 1
[0] Done
--- Reuse same instance ---
[0] Fresh start
[1] Nested again
[0] End
How it works: indent() increments the level and returns self, so with logger.indent() enters the context manager on the same object. __exit__ decrements the level when the with block ends. Because the level is instance state, nesting works naturally — each with logger.indent() pushes one level, and each exit pops one level.
Key insight: This context manager is reentrant (can be nested with itself) and reusable (can be used in multiple sequential with blocks). Not all context managers have these properties — @contextmanager-based ones are typically single-use because generator functions cannot be restarted after they finish. Class-based context managers are reusable by default if their state resets properly.
class IndentLogger:
# Implement a reentrant context manager that tracks indent level
# __init__: set level to 0
# __enter__: return self (do NOT increment here — see log method)
# __exit__: decrement level
# log(message): print with indentation based on current level
# indent(): increment level and return self (called in nested with)
pass
# Test: create one logger, use it at multiple nesting levels
# Then reuse the same instance in a new with block
Expected Output
--- Reusable (IndentLogger) ---
[0] Starting
[1] Level 1
[2] Level 2
[1] Back to 1
[0] Done
--- Reuse same instance ---
[0] Fresh start
[1] Nested again
[0] EndHints
Hint 1: A reusable context manager can be used in multiple with statements, one at a time.
Hint 2: A reentrant context manager can be nested — used in multiple with statements simultaneously.
Hint 3: __enter__ should increment the indent level, __exit__ should decrement it.
Hint 4: Store the indent level as instance state so nesting works naturally.
Use contextlib.suppress() in three scenarios: loading a config with a fallback, deleting a file that may not exist, and safely processing a list with potential division errors.
import os
from contextlib import suppress
# Task 1: Config with fallback
config_value = "default_value"
with suppress(FileNotFoundError):
with open("missing_config.ini") as f:
config_value = f.read().strip()
print(f"Config loaded: {config_value}")
# Task 2: Safe file deletion
with open("temp.txt", "w") as f:
f.write("temporary")
os.remove("temp.txt")
print("Deleted existing file")
with suppress(FileNotFoundError):
os.remove("temp.txt")
print("Delete of non-existent file: no error")
# Task 3: Safe division
divisors = [1, 0, 2, 0, 4]
results = []
for d in divisors:
with suppress(ZeroDivisionError):
results.append(10.0 / d)
continue
results.append("skip")
print(f"Results: {results}")Solution
import os
from contextlib import suppress
# Task 1: Config with fallback
config_value = "default_value"
with suppress(FileNotFoundError):
with open("missing_config.ini") as f:
config_value = f.read().strip()
print(f"Config loaded: {config_value}")
# Task 2: Safe file deletion
with open("temp.txt", "w") as f:
f.write("temporary")
os.remove("temp.txt")
print("Deleted existing file")
with suppress(FileNotFoundError):
os.remove("temp.txt")
print("Delete of non-existent file: no error")
# Task 3: Safe division
divisors = [1, 0, 2, 0, 4]
results = []
for d in divisors:
with suppress(ZeroDivisionError):
results.append(10.0 / d)
continue
results.append("skip")
print(f"Results: {results}")
Output:
Config loaded: default_value
Deleted existing file
Delete of non-existent file: no error
Results: [10.0, 'skip', 5.0, 'skip', 2.5]
How it works: suppress(ExceptionType) catches and silences the specified exception inside its with block. In Task 1, if the file does not exist, the FileNotFoundError is suppressed and config_value keeps its default. In Task 2, the second os.remove() fails silently. In Task 3, when d is 0, the ZeroDivisionError is suppressed, the continue never executes, and "skip" is appended instead.
Key insight: suppress is syntactically cleaner than try/except: pass for simple cases. However, do not use it when you need to log the error, transform the exception, or take corrective action — use explicit try/except for those scenarios.
import os
from contextlib import suppress
# Task 1: Try to read a config file that does not exist.
# Use suppress(FileNotFoundError) — if the file is missing,
# fall back to "default_value"
# Task 2: Delete a file if it exists, using suppress(FileNotFoundError)
# First create "temp.txt", delete it, then try deleting again
# Task 3: Safe division — process a list of divisors,
# skip any that cause ZeroDivisionError
Expected Output
Config loaded: default_value
Deleted existing file
Delete of non-existent file: no error
Results: [10.0, 'skip', 5.0, 'skip', 2.5]Hints
Hint 1: contextlib.suppress(*exceptions) silences the specified exceptions inside the with block.
Hint 2: It is equivalent to a try/except that passes — but more readable.
Hint 3: You can suppress multiple exception types at once.
Hint 4: Use it for optional operations where failure is acceptable.
Hard
Use contextlib.ExitStack to dynamically open a variable number of files, write to all of them, verify they are all closed after the block, and read them back.
from contextlib import ExitStack
filenames = ["out_0.txt", "out_1.txt", "out_2.txt"]
with ExitStack() as stack:
files = [stack.enter_context(open(fn, "w")) for fn in filenames]
print(f"Opened {len(files)} files")
for f in files:
f.write("hello")
print(f"Written to {len(files)} files")
all_closed = all(f.closed for f in files)
print(f"All files closed: {all_closed}")
parts = []
for i, fn in enumerate(filenames):
with open(fn) as f:
parts.append(f"file_{i}:{f.read()}")
print(f"Combined: {' '.join(parts)}")Solution
from contextlib import ExitStack
filenames = ["out_0.txt", "out_1.txt", "out_2.txt"]
with ExitStack() as stack:
files = [stack.enter_context(open(fn, "w")) for fn in filenames]
print(f"Opened {len(files)} files")
for f in files:
f.write("hello")
print(f"Written to {len(files)} files")
all_closed = all(f.closed for f in files)
print(f"All files closed: {all_closed}")
parts = []
for i, fn in enumerate(filenames):
with open(fn) as f:
parts.append(f"file_{i}:{f.read()}")
print(f"Combined: {' '.join(parts)}")
Output:
Opened 3 files
Written to 3 files
All files closed: True
Combined: file_0:hello file_1:hello file_2:hello
How it works: ExitStack maintains an internal stack of cleanup callbacks. stack.enter_context(cm) calls cm.__enter__(), stores the cm.__exit__ method on the stack, and returns the result. When the ExitStack's with block exits, it pops and calls each __exit__ in LIFO (last-in-first-out) order. This is essential when you do not know at code-writing time how many context managers you need.
Key insight: You cannot write with open(f) for f in filenames — the with statement requires a fixed number of context managers known at compile time. ExitStack solves this by letting you register them dynamically in a loop. If opening the third file fails, the first two are still properly closed.
from contextlib import ExitStack
filenames = ["out_0.txt", "out_1.txt", "out_2.txt"]
# Use ExitStack to open all files dynamically
# Write "hello" to each file
# After the with block, verify all files are closed
# Then read them all back and print combined content
Expected Output
Opened 3 files
Written to 3 files
All files closed: True
Combined: file_0:hello file_1:hello file_2:helloHints
Hint 1: ExitStack lets you register context managers dynamically in a loop.
Hint 2: Use stack.enter_context(cm) to add a context manager to the stack.
Hint 3: All registered context managers are cleaned up when the ExitStack exits.
Hint 4: ExitStack itself is a context manager — use it with a with statement.
Implement an audit_operation(operation, user) context manager using @contextmanager that logs entry, success, or failure. On failure, it should log the error details and re-raise the exception.
from contextlib import contextmanager
@contextmanager
def audit_operation(operation, user):
print(f"[audit] ENTER operation={operation}, user={user}")
try:
yield
except Exception as e:
print(f"[audit] FAILURE operation={operation}, user={user}, error={e}")
raise
else:
print(f"[audit] SUCCESS operation={operation}, user={user}")
with audit_operation("deploy", "alice"):
pass
print("Deploy finished")
try:
with audit_operation("rollback", "bob"):
raise IOError("Disk full")
except IOError as e:
print(f"Caught: {e}")Solution
from contextlib import contextmanager
@contextmanager
def audit_operation(operation, user):
print(f"[audit] ENTER operation={operation}, user={user}")
try:
yield
except Exception as e:
print(f"[audit] FAILURE operation={operation}, user={user}, error={e}")
raise
else:
print(f"[audit] SUCCESS operation={operation}, user={user}")
with audit_operation("deploy", "alice"):
pass
print("Deploy finished")
try:
with audit_operation("rollback", "bob"):
raise IOError("Disk full")
except IOError as e:
print(f"Caught: {e}")
Output:
[audit] ENTER operation=deploy, user=alice
[audit] SUCCESS operation=deploy, user=alice
Deploy finished
[audit] ENTER operation=rollback, user=bob
[audit] FAILURE operation=rollback, user=bob, error=Disk full
Caught: Disk full
How it works: The try/except/else pattern around yield gives three distinct paths: (1) except handles exceptions from inside the with block, (2) else runs only if no exception occurred, (3) finally (not used here) would run regardless. The raise in the except block re-raises the original exception so calling code can still handle it.
Key insight: The else clause on a try block runs only when no exception was raised. This is cleaner than putting success logic after yield (which would run whether or not an exception occurred if you had a bare except that did not re-raise). The try/except/else pattern makes the success and failure paths explicit and separate.
from contextlib import contextmanager
# Implement an audit_operation(operation, user) context manager
# On enter: print "[audit] ENTER operation=..., user=..."
# On success: print "[audit] SUCCESS operation=..., user=..."
# On failure: print "[audit] FAILURE operation=..., user=..., error=..."
# then re-raise the exception
# Test 1: successful operation
# Test 2: failed operation (catch the re-raised exception outside)
Expected Output
[audit] ENTER operation=deploy, user=alice
[audit] SUCCESS operation=deploy, user=alice
Deploy finished
[audit] ENTER operation=rollback, user=bob
[audit] FAILURE operation=rollback, user=bob, error=Disk full
Caught: Disk fullHints
Hint 1: Use @contextmanager — code before yield is enter, after yield is exit.
Hint 2: Wrap yield in try/except/else to handle both success and failure paths.
Hint 3: In the except block, log the failure and re-raise with raise.
Hint 4: In the else block (no exception), log success.
Use ExitStack with both context managers and callbacks to demonstrate LIFO cleanup ordering. Register resources and a callback function, then observe the reverse-order cleanup.
from contextlib import ExitStack, contextmanager
class Resource:
def __init__(self, name):
self.name = name
def __enter__(self):
print(f"Acquired resource: {self.name}")
return self
def __exit__(self, exc_type, exc_val, exc_tb):
print(f"Closing: {self.name}")
return False
def flush_logs():
print("Flushing logs to disk...")
print("--- Setup phase ---")
with ExitStack() as stack:
db = stack.enter_context(Resource("database"))
cache = stack.enter_context(Resource("cache"))
stack.callback(flush_logs)
print("Registered callback: flush_logs")
metrics = stack.enter_context(Resource("metrics"))
print("--- Work phase ---")
resources = [db, cache, metrics]
print(f"All {len(resources)} resources active")
print("--- Cleanup phase (automatic, LIFO order) ---")
print("--- All cleaned up ---")Solution
from contextlib import ExitStack, contextmanager
class Resource:
def __init__(self, name):
self.name = name
def __enter__(self):
print(f"Acquired resource: {self.name}")
return self
def __exit__(self, exc_type, exc_val, exc_tb):
print(f"Closing: {self.name}")
return False
def flush_logs():
print("Flushing logs to disk...")
print("--- Setup phase ---")
with ExitStack() as stack:
db = stack.enter_context(Resource("database"))
cache = stack.enter_context(Resource("cache"))
stack.callback(flush_logs)
print("Registered callback: flush_logs")
metrics = stack.enter_context(Resource("metrics"))
print("--- Work phase ---")
resources = [db, cache, metrics]
print(f"All {len(resources)} resources active")
print("--- Cleanup phase (automatic, LIFO order) ---")
print("--- All cleaned up ---")
Output:
--- Setup phase ---
Acquired resource: database
Acquired resource: cache
Registered callback: flush_logs
Acquired resource: metrics
--- Work phase ---
All 3 resources active
--- Cleanup phase (automatic, LIFO order) ---
Closing: metrics
Flushing logs to disk...
Closing: cache
Closing: database
--- All cleaned up ---
How it works: ExitStack maintains an internal stack (hence the name). Each enter_context() pushes a __exit__ method, and each callback() pushes a plain function. When the ExitStack's own __exit__ runs, it pops the stack in LIFO order: metrics.__exit__ first (registered last), then flush_logs (registered third), then cache.__exit__ (registered second), then database.__exit__ (registered first).
Key insight: The LIFO ordering is intentional and matches how nested with statements unwind. Resources acquired later often depend on resources acquired earlier (e.g., a database cursor depends on a connection), so they must be cleaned up first. ExitStack.callback() lets you interleave arbitrary cleanup logic at any point in the stack — the callback runs at the position where it was registered, not at the end.
Production use case: ExitStack is invaluable in setup functions where you acquire multiple resources conditionally. If the third resource fails to acquire, the first two are still cleaned up properly. Without ExitStack, you would need deeply nested with statements or complex try/finally chains.
from contextlib import ExitStack, contextmanager
# Implement a Resource class that is a context manager
# __enter__ prints "Acquired resource: {name}" and returns self
# __exit__ prints "Closing: {name}"
# Implement a flush_logs() function that prints "Flushing logs to disk..."
# Use ExitStack to:
# 1. Enter resource "database"
# 2. Enter resource "cache"
# 3. Register flush_logs as a callback
# 4. Enter resource "metrics"
# Observe the LIFO cleanup order
Expected Output
--- Setup phase ---
Acquired resource: database
Acquired resource: cache
Registered callback: flush_logs
Acquired resource: metrics
--- Work phase ---
All 3 resources active
--- Cleanup phase (automatic, LIFO order) ---
Closing: metrics
Flushing logs to disk...
Closing: cache
Closing: database
--- All cleaned up ---Hints
Hint 1: ExitStack.enter_context() registers a context manager for cleanup.
Hint 2: ExitStack.callback(func, *args) registers a plain function to call on exit.
Hint 3: Cleanup happens in LIFO order — last registered, first cleaned up.
Hint 4: Callbacks and context manager exits are interleaved based on registration order.