Step 07 of 13
Exceptions & context managers
Design custom exceptions, chain and annotate errors, use try/except/else/finally correctly, and write context managers that always clean up.
Beginner code catches ValueError from int(). Production code has to decide which errors a module raises, how callers tell them apart, how to keep the original cause when wrapping an error, and how to guarantee that files, locks and connections are released whatever goes wrong. This lesson covers designing exceptions, the full try statement, exception chaining and groups, and context managers, the with statement's machinery for reliable cleanup.
The exception hierarchy
Exceptions are classes arranged in a hierarchy, and except SomeError catches that class and its subclasses:
BaseException
├── KeyboardInterrupt, SystemExit (don't catch these in normal code)
└── Exception
├── ValueError
│ └── UnicodeDecodeError, json.JSONDecodeError
├── LookupError
│ ├── KeyError
│ └── IndexError
├── OSError
│ ├── FileNotFoundError
│ ├── PermissionError
│ └── TimeoutError
└── ...So except LookupError catches both KeyError and IndexError, and except OSError catches every file and network error. Catch the narrowest class that covers what you can actually handle.
except Exception catches nearly everything. It's legitimate in exactly one kind of place: the top level of a program or request handler, where you log the error and keep the server alive. Anywhere else it hides bugs.
Designing your own exceptions
A module or library should raise exceptions that mean something in its own terms. Define a base class for the package and specific subclasses:
class PaymentError(Exception):
"""Base class for errors raised by the payments module."""
class CardDeclined(PaymentError):
def __init__(self, reason: str) -> None:
super().__init__(f"card declined: {reason}")
self.reason = reason
class ProviderUnavailable(PaymentError):
"""The payment provider could not be reached; safe to retry later."""Callers can then handle precisely what they care about:
try:
charge(order)
except CardDeclined as e:
show_message(f"Your card was declined ({e.reason}).")
except ProviderUnavailable:
schedule_retry(order)
except PaymentError:
alert_on_call("unexpected payment failure")
raiseGuidelines:
- Raise early, with a clear message that includes the offending value:
ValueError(f"quantity must be positive, got {qty}"). - Don't invent exceptions the standard library already has. Bad arguments are
ValueErrororTypeError; missing keys areKeyError. - Attach data as attributes (
e.reason), so callers don't parse message strings. - A bare
raiseinsideexceptre-raises the current exception with its original traceback, after you've logged or cleaned up.
try, except, else, finally
The full statement has four parts:
def read_config(path: str) -> dict:
try:
f = open(path, encoding="utf-8")
except FileNotFoundError:
return {} # expected: first run, no config yet
else:
with f:
return json.load(f) # runs only if open() succeeded
finally:
print(f"config lookup for {path} finished") # always runsexceptruns if thetryblock raised a matching exception.elseruns if thetryblock didn't raise. Putting follow-up code here, rather than insidetry, means aJSONDecodeErrorfromjson.loadisn't accidentally caught by anexceptmeant foropen.finallyruns no matter what: success, exception orreturn. It's for cleanup, though as you'll see, a context manager is usually a better way to clean up.
Chaining: keep the original cause
When you translate a low-level error into your own, keep the original with raise ... from:
import httpx
def fetch_rates() -> dict[str, float]:
try:
response = httpx.get("https://rates.example.com/latest", timeout=5)
response.raise_for_status()
except httpx.TransportError as e:
raise ProviderUnavailable("exchange rate service unreachable") from e
return response.json()["rates"]The traceback then shows both, joined by "The above exception was the direct cause of the following exception", so whoever debugs it sees the real network error behind your friendly one. Use raise NewError(...) from None only when the original is genuinely irrelevant noise.
Python 3.11 added add_note, which attaches context without changing the exception type:
for line_no, line in enumerate(lines, start=1):
try:
records.append(parse(line))
except ValueError as e:
e.add_note(f"while parsing line {line_no} of {path}")
raiseThe note appears at the end of the traceback, which is often exactly the missing clue.
Exception groups
Sometimes several things fail at once, such as concurrent tasks or validating many fields. Python 3.11 added ExceptionGroup to raise them together, and except* to handle each kind:
def validate(user: dict) -> None:
errors: list[Exception] = []
if "@" not in user.get("email", ""):
errors.append(ValueError("invalid email"))
if user.get("age", 0) < 0:
errors.append(ValueError("age must not be negative"))
if "name" not in user:
errors.append(KeyError("name"))
if errors:
raise ExceptionGroup("invalid user", errors)
try:
validate({"email": "nope", "age": -1})
except* ValueError as group:
for e in group.exceptions:
print("value problem:", e)
except* KeyError as group:
print("missing fields:", [str(e) for e in group.exceptions])value problem: invalid email
value problem: age must not be negative
missing fields: ["'name'"]Unlike except, every matching except* clause runs, each receiving the subgroup of exceptions it matches. You'll meet exception groups mostly through asyncio.TaskGroup in the async lesson.
Context managers: cleanup you can't forget
Every resource that must be released, such as files, locks, database connections and temporary directories, has the same shape: acquire, use, release, even if the use fails. The with statement packages that:
with open("data.txt", encoding="utf-8") as f:
process(f)
# f is closed here, whether process() returned normally or raisedAn object works with with if it has two methods. __enter__ runs at the start, and its return value is bound by as. __exit__ runs at the end, receiving details of any exception:
import time
class Timer:
def __enter__(self) -> "Timer":
self.start = time.perf_counter()
return self
def __exit__(self, exc_type, exc, tb) -> None:
self.elapsed = time.perf_counter() - self.start
print(f"took {self.elapsed:.3f}s")
# returning None (falsy) lets any exception propagate
with Timer() as t:
sum(range(10_000_000))If __exit__ returns True, the exception is suppressed. Do that only deliberately.
Writing one with contextlib
For most context managers, a generator is shorter. @contextmanager turns the code before yield into __enter__ and the code after into __exit__:
from collections.abc import Iterator
from contextlib import contextmanager
import os
@contextmanager
def working_directory(path: str) -> Iterator[None]:
previous = os.getcwd()
os.chdir(path)
try:
yield
finally:
os.chdir(previous) # runs even if the with-block raised
with working_directory("/tmp"):
print(os.getcwd()) # /tmp
print(os.getcwd()) # back where you wereThe try/finally around yield is essential: without it, an exception inside the with block would skip the cleanup.
contextlib has ready-made ones too:
from contextlib import ExitStack, suppress
with suppress(FileNotFoundError): # "delete if it exists"
os.remove("cache.tmp")
with ExitStack() as stack: # a variable number of resources
files = [stack.enter_context(open(p, encoding="utf-8")) for p in paths]
merge(files) # all closed on exit, in reverse orderOther standard context managers you'll use: threading.Lock (with lock:), tempfile.TemporaryDirectory(), database transactions in most drivers (with conn.transaction():), and unittest.mock.patch in tests.
Practice
Write atomic_write(path), a context manager for safely replacing a file:
- Inside the
withblock, the caller writes to a file object as usual. - The data is actually written to a temporary file in the same directory.
- If the block finishes without an exception, the temporary file replaces the target in one step (
os.replace), so readers never see a half-written file. - If the block raises, the temporary file is deleted, the original file is untouched, and the exception propagates.
Then use it to save JSON, and test that an exception halfway through leaves the old file intact.
Show solution
import json
import os
import tempfile
from collections.abc import Iterator
from contextlib import contextmanager
from typing import TextIO
@contextmanager
def atomic_write(path: str, encoding: str = "utf-8") -> Iterator[TextIO]:
directory = os.path.dirname(os.path.abspath(path))
fd, tmp_path = tempfile.mkstemp(dir=directory, prefix=".tmp-")
try:
with os.fdopen(fd, "w", encoding=encoding) as tmp:
yield tmp
tmp.flush()
os.fsync(tmp.fileno()) # make sure the bytes reached the disk
os.replace(tmp_path, path) # atomic on the same filesystem
except BaseException:
os.unlink(tmp_path)
raise
# Normal use
with atomic_write("settings.json") as f:
json.dump({"theme": "dark"}, f)
# A failure halfway leaves the old file intact
try:
with atomic_write("settings.json") as f:
f.write('{"theme": ')
raise RuntimeError("crashed mid-write")
except RuntimeError:
pass
with open("settings.json", encoding="utf-8") as f:
print(json.load(f)) # {'theme': 'dark'}The temporary file is created in the same directory as the target because os.replace is only atomic within one filesystem; a temp file in /tmp might live on a different one. Catching BaseException (not just Exception) means even Ctrl+C mid-write cleans up the temporary file, and the bare raise re-raises the original exception unchanged. The guarantee comes from os.replace: at every instant, settings.json is either the complete old version or the complete new one.
Key takeaways
exceptcatches a class and its subclasses: catch the narrowest class you can handle, andExceptiononly at top-level boundaries.- Give a module its own exception hierarchy with data in attributes, and re-raise with bare
raiseafter logging. - Use
elsefor code that should run only whentrysucceeded, andfinallyfor unconditional cleanup. - Preserve causes with
raise ... from e, add context withadd_note, and handle concurrent failures withExceptionGroupandexcept*. - Context managers pair acquire and release; write them with
@contextmanagerand atry/finallyaroundyield.
Check yourself
Quiz
0/4 correct · 4 left
1.Which clause catches both KeyError and IndexError?
2.Why put follow-up code in the else block of a try statement rather than inside try?
3.What does raise ProviderUnavailable("...") from e achieve?
4.In a @contextmanager function, why wrap yield in try/finally?
Finished reading?
Mark it complete to fill in your roadmap.