Python · lesson 24 of 34
Beginner
16- 1 Introduction to Python
- 2 Setting Up Python
- 3 Python Variables and Type System
- 4 Python Data Types
- 5 Python Operators
- 6 Python Strings
- 7 Control Flow in Python
- 8 Python Functions and Parameters
- 9 Python Data Structures
- 10 Python Collections
- 11 Python Modules and Packages
- 12 Object-Oriented Programming in Python
- 13 Encapsulation and Name Mangling in Python
- 14 Polymorphism and Duck Typing in Python
- 15 Abstraction with Abstract Base Classes in Python
- 21 Building a Python CLI Tool
Intermediate
12- 16 Multiple Inheritance and MRO in Python
- 17 Composition vs Inheritance in Python
- 18 Python Dataclasses
- 19 Error Handling in Python
- 20 File I/O, pathlib, and JSON in Python
- 22 Python Decorators and Closures
- 23 Python Generators
- 24 Python Context Managers
- 25 Python Concurrency
- 27 Testing Python Code
- 28 Python Security
- 34 Python Interview Prep
Advanced
6Python Context Managers
See exactly when __enter__ and __exit__ run, then build context managers with classes, @contextmanager, suppress and ExitStack, with real output.
By SkillByExample · Updated
A context manager pairs setup with guaranteed teardown. The with statement calls __enter__ going in and __exit__ coming out — even when the block raises. Output below is from Python 3.13.
The protocol, traced
A class that prints from both hooks shows the exact order of events, with and without an exception.
class Tracer:
def __enter__(self):
print("enter")
return "the as-value"
def __exit__(self, exc_type, exc_val, exc_tb):
print("exit:", exc_type.__name__ if exc_type else None)
return False # don't suppress
with Tracer() as value:
print("body got", value)
try:
with Tracer():
print("body raising")
raise ValueError("boom")
print("never printed")
except ValueError as e:
print("caller caught:", e)
$ python tracer.py
enter
body got the as-value
exit: None
enter
body raising
exit: ValueError
caller caught: boom
Whatever __enter__ returns is what as binds — not the Tracer object itself. __exit__ ran before the caller’s except, received the exception’s type, and returned False, so the ValueError continued outward.
Files: the context manager you already use
open() returns a file object whose __exit__ closes it.
with open("data.txt", "w") as f:
f.write("hello\n")
print(f.closed)
try:
with open("data.txt") as f:
line = f.readline()
raise RuntimeError("processing failed")
except RuntimeError:
pass
print(repr(line), f.closed)
with open("data.txt") as src, open("copy.txt", "w") as dst:
dst.write(src.read().upper())
print(open("copy.txt").read(), end="")
$ python files.py
True
'hello\n' True
HELLO
The file was closed in both cases — including after the RuntimeError. The two-manager form is the same as nesting two with blocks; dst closes first, then src.
Suppressing an exception
If __exit__ returns something truthy, the exception is swallowed.
import os
class Suppress:
def __init__(self, *exc_types):
self.exc_types = exc_types
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
return exc_type is not None and issubclass(exc_type, self.exc_types)
with Suppress(FileNotFoundError):
os.remove("does-not-exist.tmp")
print("not reached")
print("carried on")
with Suppress(FileNotFoundError):
int("abc")
$ python suppress_class.py
carried on
Traceback (most recent call last):
File "/home/user/py-demo/context-managers/suppress_class.py", line 19, in <module>
int("abc")
~~~^^^^^^^
ValueError: invalid literal for int() with base 10: 'abc'
The rest of the block after the failing line is skipped — suppression resumes after the with, not after the line. Only the listed types are swallowed; the ValueError still escaped. The standard library already ships this as contextlib.suppress.
@contextmanager: a generator instead of a class
Code before yield is __enter__, the yielded value is the as target, and code after it is __exit__. A try/finally makes the teardown run on errors too.
import os
import tempfile
from contextlib import contextmanager
@contextmanager
def working_directory(path: str):
original = os.getcwd()
os.chdir(path)
try:
yield path
finally:
os.chdir(original)
start = os.getcwd()
tmp = os.path.realpath(tempfile.mkdtemp())
with working_directory(tmp) as p:
print(os.getcwd() == p)
try:
with working_directory(tmp):
raise KeyError("oops")
except KeyError:
pass
print(os.getcwd() == start)
$ python chdir.py
True
True
Even after the KeyError, the directory was restored, because the exception is thrown into the generator at the yield and the finally runs. Drop the try/finally and an error would leave the process in the wrong directory. (Python 3.11 added contextlib.chdir, which does exactly this.)
A transaction: commit or roll back
Yielding a value and re-raising in except gives commit-on-success, rollback-on-error.
import sqlite3
from contextlib import contextmanager
conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE stock (item TEXT, qty INTEGER CHECK (qty >= 0))")
conn.execute("INSERT INTO stock VALUES ('book', 1)")
conn.commit()
@contextmanager
def transaction(connection):
try:
yield connection.cursor()
connection.commit()
except Exception:
connection.rollback()
raise
for attempt in (1, 2):
try:
with transaction(conn) as cur:
cur.execute("INSERT INTO stock VALUES ('pen', 5)")
cur.execute("UPDATE stock SET qty = qty - 1 WHERE item = 'book'")
except sqlite3.IntegrityError as e:
print(f"attempt {attempt} rolled back:", e)
print(conn.execute("SELECT * FROM stock").fetchall())
$ python transaction.py
[('book', 0), ('pen', 5)]
attempt 2 rolled back: CHECK constraint failed: qty >= 0
[('book', 0), ('pen', 5)]
The second attempt’s UPDATE would take the book stock to -1, so the constraint fails — and the rollback also undoes that attempt’s INSERT, which had already succeeded. Only one pen row exists.
contextlib utilities
suppress and redirect_stdout cover two common cases without writing a class.
import io
import os
from contextlib import redirect_stdout, suppress
with suppress(FileNotFoundError):
os.remove("temp.txt")
buffer = io.StringIO()
with redirect_stdout(buffer):
print("captured line")
print("another")
print("after:", buffer.getvalue().splitlines())
$ python contextlib_utils.py
after: ['captured line', 'another']
Inside the redirect_stdout block, print wrote to the StringIO instead of the terminal; on exit sys.stdout was restored.
ExitStack: a runtime number of managers
When you don’t know until runtime how many files to open, ExitStack collects them and closes them all.
from contextlib import ExitStack
names = ["a.txt", "b.txt", "c.txt"]
for n in names:
with open(n, "w") as f:
f.write(n.upper())
with ExitStack() as stack:
handles = [stack.enter_context(open(n)) for n in names]
stack.callback(print, "stack unwinding")
print([h.read() for h in handles])
print([h.closed for h in handles])
$ python exitstack.py
['A.TXT', 'B.TXT', 'C.TXT']
stack unwinding
[True, True, True]
callback registers a plain function as a teardown step. Teardown runs last-in first-out, so the callback (registered last) ran before the files were closed.
Lock vs RLock as context managers
Locks are context managers too. A plain Lock can’t be re-acquired by the thread that holds it; an RLock can.
import threading
lock = threading.Lock()
rlock = threading.RLock()
with lock:
print("Lock again:", lock.acquire(timeout=0.1)) # would deadlock without timeout
with rlock:
with rlock:
print("RLock nested OK")
$ python locks.py
Lock again: False
RLock nested OK
acquire(timeout=0.1) gave up and returned False; a plain nested with lock: would hang forever. An RLock counts acquisitions by the owning thread and releases only when the count returns to zero.
Practice
1. Write a Timer class-based context manager that stores the elapsed time on itself, and show it still records a value when the block raises.
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.elapsed = time.perf_counter() - self.start
return False
try:
with Timer() as t:
time.sleep(0.2)
raise RuntimeError("fail")
except RuntimeError:
pass
print(f"{t.elapsed:.1f}s")
$ python timer.py
0.2s
__exit__ runs before the exception propagates, so elapsed is set either way. The exact figure varies slightly run to run.
2. Rewrite Timer as a @contextmanager generator that prints the elapsed time when the block ends.
Solution
import time
from contextlib import contextmanager
@contextmanager
def timer(label: str):
start = time.perf_counter()
try:
yield
finally:
print(f"{label}: {time.perf_counter() - start:.1f}s")
with timer("sleep"):
time.sleep(0.3)
$ python timer_gen.py
sleep: 0.3s
The finally is what makes it print even if the block raises.
3. Write override(obj, **attrs) with @contextmanager that temporarily sets attributes on an object and restores the old values afterwards.
Solution
from contextlib import contextmanager
class Config:
debug = False
level = "INFO"
@contextmanager
def override(obj, **attrs):
saved = {k: getattr(obj, k) for k in attrs}
for k, v in attrs.items():
setattr(obj, k, v)
try:
yield obj
finally:
for k, v in saved.items():
setattr(obj, k, v)
with override(Config, debug=True, level="DEBUG"):
print(Config.debug, Config.level)
print(Config.debug, Config.level)
$ python override.py
True DEBUG
False INFO
4. Use ExitStack to acquire a lock only when a flag is set, then show both paths.
Solution
import threading
from contextlib import ExitStack
lock = threading.Lock()
def work(need_lock: bool) -> None:
with ExitStack() as stack:
if need_lock:
stack.enter_context(lock)
print(f"need_lock={need_lock}, locked={lock.locked()}")
print("after:", lock.locked())
work(True)
work(False)
$ python conditional_lock.py
need_lock=True, locked=True
after: False
need_lock=False, locked=False
after: False
An empty ExitStack is a no-op, so the conditional manager needs no duplicated with blocks.