A function reads a report. Sometimes the caller passes in a stream that is already open, and the function must leave it open, because the caller still owns it. Other times the function opens the stream itself, and then it must close it. If you write the reading code twice, once per case, a later fix can change one copy and miss the other.
Two helpers from contextlib let you choose the context manager in one
expression and write the reading code once:
from collections.abc import Callable
from contextlib import AbstractContextManager, closing, nullcontext
from io import StringIO
def read_text(stream: StringIO | None, open_new: Callable[[], StringIO]) -> str:
scope: AbstractContextManager[StringIO] = (
nullcontext(stream) if stream is not None else closing(open_new())
)
with scope as active:
return active.read()
given = StringIO("borrowed")
print(read_text(given, lambda: StringIO("unused"))) # borrowed
print(given.closed) # False
created = StringIO("owned")
print(read_text(None, lambda: created)) # owned
print(created.closed) # True
nullcontext gives you the object and does nothing else
nullcontext(x) is a context manager that gives you x on entry and does
nothing on exit. Use it when you borrow a resource. The with statement
still works, and nothing gets closed.
closing calls close() when the block ends
closing(x) gives you x on entry and calls x.close() on exit, even when
the block raises. Use it for an object that has a close() method but is not
a context manager. StringIO and files already work with with, so for a
real file the owned case can be open(path, encoding="utf-8") with no
closing.
The annotation AbstractContextManager[StringIO] tells mypy that both
choices are context managers that give you a StringIO. Then mypy knows the
type of active.
suppress skips one expected error
suppress(SomeError) stops that exception type at the end of the with
block. It works like try with except SomeError: pass. Python skips the
rest of the block and continues after it. It does not go back to the line
after the one that failed.
from contextlib import suppress
values: dict[str, int] = {}
events: list[str] = []
with suppress(KeyError):
value = values["missing"]
events.append("not reached")
events.append("after block")
print(events) # ['after block']
A common use is a default for a missing file:
def read_or_default(load: Callable[[], str], default: str) -> str:
with suppress(FileNotFoundError):
return load()
return default
def missing() -> str:
raise FileNotFoundError("settings.txt")
print(read_or_default(lambda: "payload", "fallback")) # payload
print(read_or_default(missing, "fallback")) # fallback
A common mistake is suppress(Exception). It would also hide a
PermissionError, or a bug inside load. Your program would then continue
with the default as if the file were missing. Name the one error you expect.
Keep the block small too, because every line after the failing one is
skipped.
Check your understanding
Why does read_or_default catch only FileNotFoundError?
Show answer
Hide answer
A missing file is the one failure where the default is the right answer. A
PermissionError means the file exists but the program cannot read it. That
is usually a deployment mistake. If you hide it, the service quietly
runs on its default settings. Let it propagate so that someone sees it and
fixes the permissions.
chdir changes the working directory for one block
contextlib.chdir(path) changes the working directory and changes it back
when the block ends, even if the block raises:
import os
import tempfile
from contextlib import chdir
before = os.getcwd()
with tempfile.TemporaryDirectory() as tmp:
with chdir(tmp):
print(os.getcwd() == before) # False
print(os.getcwd() == before) # True
The working directory belongs to the whole process, so any other code that
runs during the block sees the change too. chdir is handy in scripts and
tests. In library code, pass full paths instead.
Going further
nullcontext()with no argument gives youNone.- Since Python 3.12,
suppressalso handles exception groups. It removes the exceptions that match and raises the rest again. You meet exception groups with asyncio'sTaskGroupin the concurrency track. - See the contextlib documentation.