Warm-up · Activity 1 of 7
// I1.5 · ~40 min · Intermediate
Build: a retrying file reader with contextlib
In this build you write your own context managers with @contextmanager, use suppress and closing, and combine this module into a file reader that retries, reports every failed attempt and always removes its lock file.
Lesson 5 of 5 in I1 Errors in depth
PreviousEnd of the module
You will be able to
- Write a context manager with @contextmanager, with try and finally around the yield
- Use suppress to ignore one expected exception and closing to call close() for you
- Build a retry loop that notes and collects each failure and raises one clear error from the group
Predict · Activity 2 of 7
Predict before you read on: a context manager made from a function with yield, and a block that fails. What does this print?
from contextlib import contextmanager @contextmanager def step(name): print("start", name, end=" | ") yield print("end", name, end=" | ") try: with step("load"): raise OSError("disk") except OSError: print("caught")Practice · Activity 3 of 7
Two new pieces of syntax in step above: @contextmanager on the line above def, and yield inside it. Which statements are true? Pick all that apply.
Practice · Activity 4 of 7
The lock file does not exist. What does this print?
import os from contextlib import suppress with suppress(FileNotFoundError): os.remove("no-such.lock") print("removed", end=" ") print("done")Practice · Activity 5 of 7
Match each tool to what it does for you.
Brain teaser · Activity 6 of 7
Brain teaser. missing.txt does not exist. What does this print?
from contextlib import contextmanager @contextmanager def attempt(n): try: yield except OSError: print("attempt", n, "failed", end=" | ") for n in [1, 2]: with attempt(n): open("missing.txt") print("read ok", end=" | ") print("after")Apply · Activity 7 of 7
Mini-task. Write a context manager temporarily_set(settings, key, value) with @contextmanager. It sets settings[key] = value for the with block and puts the old value back afterwards, even when the block raises. If the key did not exist before, it is removed again. Try it with a block that raises and with a new key.
Check your work against this list
Build it yourself
Read the worked example, then write the exercises. Your code runs in your browser or on your computer and is never uploaded.
Worked example
A cleanup step with all three tools
step is a context manager written with @contextmanager: it records start, failure and end in a report, and re-raises the error. suppress around it ignores the FileNotFoundError when old.lock is already gone, and closing calls report.close() at the end, because Report has close() but no with support of its own.
main.py
import os
from collections.abc import Iterator
from contextlib import closing, contextmanager, suppress
class Report:
def __init__(self, name: str) -> None:
self.name = name
self.lines: list[str] = []
def add(self, line: str) -> None:
self.lines.append(line)
def close(self) -> None:
print(f"{self.name}: {len(self.lines)} lines, closed")
@contextmanager
def step(report: Report, name: str) -> Iterator[None]:
report.add(f"start {name}")
try:
yield
except OSError as err:
report.add(f"failed {name}: {type(err).__name__}")
raise
finally:
report.add(f"end {name}")
with closing(Report("cleanup")) as report:
with suppress(FileNotFoundError):
with step(report, "remove old.lock"):
os.remove("old.lock")
for line in report.lines:
print(line)
Run it with
python main.pyOutput
start remove old.lock
failed remove old.lock: FileNotFoundError
end remove old.lock
cleanup: 3 lines, closed- step re-raises after logging, so the error still reaches suppress, which then ignores it.
- finally adds the end line whether os.remove works or fails.
- closing(Report(...)) binds the report itself to report, and calls close() after the loop.
- The for loop runs inside closing, so the report is printed before close() adds its own line.
Change it and run it
Tab indents and Shift+Tab outdents. To leave the editor with the keyboard, press Esc, then Tab.
The first run downloads Python for your browser (up to 6.5 MB) and keeps it cached. Your code stays on your device.
Exercises
Exercise 1 of 3
Step 1: a context manager for one attempt
Start the reader with attempt(log, number), a context manager written with @contextmanager. It appends "attempt <n>: start" on entry and "attempt <n>: end" on exit, also when the block fails. If the block raises an OSError, it appends "attempt <n>: <type name>" before the end line and re-raises the error. Other exceptions pass through unlogged, but still get the end line.
Tab indents and Shift+Tab outdents. To leave the editor with the keyboard, press Esc, then Tab.
The first run downloads Python for your browser (up to 6.5 MB) and keeps it cached. Your code stays on your device.
Hints
Hint 1
Put the yield inside try, and the end line in a finally clause.
Hint 2
Between them, add except OSError as err: that appends f"attempt {number}: {type(err).__name__}".
Hint 3
End that except block with a bare raise, or the error is swallowed.
Show a solution
One way to solve it. Yours can look different and still pass the checks.
from collections.abc import Iterator
from contextlib import contextmanager
@contextmanager
def attempt(log: list[str], number: int) -> Iterator[None]:
log.append(f"attempt {number}: start")
try:
yield
except OSError as err:
log.append(f"attempt {number}: {type(err).__name__}")
raise
finally:
log.append(f"attempt {number}: end")
Run it on your computer
Install Python 3.14 or newer. Save these files in one folder, open a terminal in that folder, and run the commands below.
main.py
from collections.abc import Iterator
from contextlib import contextmanager
@contextmanager
def attempt(log: list[str], number: int) -> Iterator[None]:
log.append(f"attempt {number}: start")
yield
log.append(f"attempt {number}: end")
test_main.py
from main import attempt
def test_success():
"""A block that works logs start and end"""
log = []
with attempt(log, 1):
pass
assert log == ["attempt 1: start", "attempt 1: end"], f"the log is {log!r}"
def test_os_error_logged_and_raised():
"""An OSError is logged with its type and re-raised"""
log = []
try:
with attempt(log, 2):
open("no_such_file.txt", encoding="utf-8")
except FileNotFoundError:
pass
else:
assert False, "the FileNotFoundError did not reach the caller; re-raise it"
assert log == ["attempt 2: start", "attempt 2: FileNotFoundError", "attempt 2: end"], f"the log is {log!r}"
def test_other_error_passes():
"""Other errors are not logged as failures but still end the attempt"""
log = []
try:
with attempt(log, 3):
int("x")
except ValueError:
pass
else:
assert False, "the ValueError did not reach the caller"
assert log == ["attempt 3: start", "attempt 3: end"], f"the log is {log!r}"
On macOS and Linux, type python3 wherever these commands say python, as in the first lesson.
Run the program:
python main.pyRun the checks (needs learnrun.py in the same folder):
python learnrun.py testExercise 2 of 3
Step 2: retry and collect
Write read_with_retry(path, attempts=3, log=None, wait=None). Try up to attempts times, each inside with attempt(log, n), n from 1, to open the file and return its text. For each OSError, add the note "attempt <n>" and collect it; call wait(n) between attempts if wait is given. If all fail, raise ReadFailedError(f"could not read {path} after {attempts} attempts") from ExceptionGroup("read errors", errors). Step 1 is in main.py. The hint Callable[[int], None] | None means a function that takes an int, or None.
Tab indents and Shift+Tab outdents. To leave the editor with the keyboard, press Esc, then Tab.
The first run downloads Python for your browser (up to 6.5 MB) and keeps it cached. Your code stays on your device.
Hints
Hint 1
Loop with for n in range(1, attempts + 1):. Inside, try: with attempt(log, n): and inside that with open(...) as f: return f.read().
Hint 2
In except OSError as err:, call err.add_note(f"attempt {n}"), append err, and call wait(n) only if wait is not None and n < attempts.
Hint 3
After the loop, raise ReadFailedError(...) from ExceptionGroup("read errors", errors). Set log = [] first if it is None.
Show a solution
One way to solve it. Yours can look different and still pass the checks.
from collections.abc import Callable, Iterator
from contextlib import contextmanager
@contextmanager
def attempt(log: list[str], number: int) -> Iterator[None]:
log.append(f"attempt {number}: start")
try:
yield
except OSError as err:
log.append(f"attempt {number}: {type(err).__name__}")
raise
finally:
log.append(f"attempt {number}: end")
class ReadFailedError(Exception):
"""The file could not be read in any attempt."""
def read_with_retry(
path: str,
attempts: int = 3,
log: list[str] | None = None,
wait: Callable[[int], None] | None = None,
) -> str:
if log is None:
log = []
errors: list[OSError] = []
for n in range(1, attempts + 1):
try:
with attempt(log, n):
with open(path, encoding="utf-8") as f:
return f.read()
except OSError as err:
err.add_note(f"attempt {n}")
errors.append(err)
if wait is not None and n < attempts:
wait(n)
raise ReadFailedError(f"could not read {path} after {attempts} attempts") from ExceptionGroup(
"read errors", errors
)
Run it on your computer
Install Python 3.14 or newer. Save these files in one folder, open a terminal in that folder, and run the commands below.
main.py
from collections.abc import Callable, Iterator
from contextlib import contextmanager
@contextmanager
def attempt(log: list[str], number: int) -> Iterator[None]:
log.append(f"attempt {number}: start")
try:
yield
except OSError as err:
log.append(f"attempt {number}: {type(err).__name__}")
raise
finally:
log.append(f"attempt {number}: end")
class ReadFailedError(Exception):
"""The file could not be read in any attempt."""
def read_with_retry(
path: str,
attempts: int = 3,
log: list[str] | None = None,
wait: Callable[[int], None] | None = None,
) -> str:
with open(path, encoding="utf-8") as f:
return f.read()
test_main.py
import os
from main import ReadFailedError, read_with_retry
def test_reads_file():
"""An existing file is read on the first attempt"""
with open("ready.txt", "w", encoding="utf-8") as f:
f.write("hello")
log = []
got = read_with_retry("ready.txt", log=log)
assert got == "hello", f"read_with_retry returned {got!r}"
assert log == ["attempt 1: start", "attempt 1: end"], f"the log is {log!r}"
def test_retries_until_file_appears():
"""The file appears during the second wait, and attempt 3 reads it"""
if os.path.exists("late.txt"):
os.remove("late.txt")
waits = []
def wait(n):
waits.append(n)
if n == 2:
with open("late.txt", "w", encoding="utf-8") as f:
f.write("ready")
log = []
got = read_with_retry("late.txt", log=log, wait=wait)
assert got == "ready", f"read_with_retry returned {got!r}"
assert waits == [1, 2], f"wait was called with {waits!r}"
assert log[-2:] == ["attempt 3: start", "attempt 3: end"], f"the log is {log!r}"
assert log.count("attempt 1: FileNotFoundError") == 1, f"the log is {log!r}"
def test_gives_up():
"""After the last attempt: ReadFailedError from a group of all errors"""
waits = []
try:
read_with_retry("never.txt", attempts=2, wait=waits.append)
except ReadFailedError as err:
assert str(err) == "could not read never.txt after 2 attempts", f"the message is {str(err)!r}"
group = err.__cause__
assert isinstance(group, ExceptionGroup), f"the cause is {group!r}, expected an ExceptionGroup"
notes = [getattr(e, "__notes__", []) for e in group.exceptions]
assert notes == [["attempt 1"], ["attempt 2"]], f"the notes are {notes!r}"
else:
assert False, "read_with_retry returned normally for a missing file"
assert waits == [1], f"wait was called with {waits!r}; no wait after the last attempt"
On macOS and Linux, type python3 wherever these commands say python, as in the first lesson.
Run the program:
python main.pyRun the checks (needs learnrun.py in the same folder):
python learnrun.py testExercise 3 of 3
Step 3: a lock file that always goes away
Write read_locked(path, attempts=3, wait=None). It creates the lock file path + ".lock" with mode "x", so it fails with FileExistsError if another reader holds the lock. Then it returns read_with_retry(path, attempts, wait=wait). Whatever happens during the read, it removes the lock afterwards; if the lock is already gone, that is fine. Steps 1 and 2 are in main.py.
Tab indents and Shift+Tab outdents. To leave the editor with the keyboard, press Esc, then Tab.
The first run downloads Python for your browser (up to 6.5 MB) and keeps it cached. Your code stays on your device.
Hints
Hint 1
Create the lock before try: if that fails, there is no lock of yours to remove.
Hint 2
Wrap return read_with_retry(...) in try, and remove the lock in the finally clause.
Hint 3
In finally, write with suppress(FileNotFoundError): os.remove(lock_path), so a lock that is already gone is fine.
Show a solution
One way to solve it. Yours can look different and still pass the checks.
import os
from collections.abc import Callable, Iterator
from contextlib import contextmanager, suppress
@contextmanager
def attempt(log: list[str], number: int) -> Iterator[None]:
log.append(f"attempt {number}: start")
try:
yield
except OSError as err:
log.append(f"attempt {number}: {type(err).__name__}")
raise
finally:
log.append(f"attempt {number}: end")
class ReadFailedError(Exception):
"""The file could not be read in any attempt."""
def read_with_retry(
path: str,
attempts: int = 3,
log: list[str] | None = None,
wait: Callable[[int], None] | None = None,
) -> str:
if log is None:
log = []
errors: list[OSError] = []
for n in range(1, attempts + 1):
try:
with attempt(log, n):
with open(path, encoding="utf-8") as f:
return f.read()
except OSError as err:
err.add_note(f"attempt {n}")
errors.append(err)
if wait is not None and n < attempts:
wait(n)
raise ReadFailedError(f"could not read {path} after {attempts} attempts") from ExceptionGroup(
"read errors", errors
)
def read_locked(path: str, attempts: int = 3, wait: Callable[[int], None] | None = None) -> str:
lock_path = path + ".lock"
with open(lock_path, "x", encoding="utf-8"):
pass
try:
return read_with_retry(path, attempts, wait=wait)
finally:
with suppress(FileNotFoundError):
os.remove(lock_path)
Run it on your computer
Install Python 3.14 or newer. Save these files in one folder, open a terminal in that folder, and run the commands below.
main.py
import os
from collections.abc import Callable, Iterator
from contextlib import contextmanager, suppress
@contextmanager
def attempt(log: list[str], number: int) -> Iterator[None]:
log.append(f"attempt {number}: start")
try:
yield
except OSError as err:
log.append(f"attempt {number}: {type(err).__name__}")
raise
finally:
log.append(f"attempt {number}: end")
class ReadFailedError(Exception):
"""The file could not be read in any attempt."""
def read_with_retry(
path: str,
attempts: int = 3,
log: list[str] | None = None,
wait: Callable[[int], None] | None = None,
) -> str:
if log is None:
log = []
errors: list[OSError] = []
for n in range(1, attempts + 1):
try:
with attempt(log, n):
with open(path, encoding="utf-8") as f:
return f.read()
except OSError as err:
err.add_note(f"attempt {n}")
errors.append(err)
if wait is not None and n < attempts:
wait(n)
raise ReadFailedError(f"could not read {path} after {attempts} attempts") from ExceptionGroup(
"read errors", errors
)
def read_locked(path: str, attempts: int = 3, wait: Callable[[int], None] | None = None) -> str:
lock_path = path + ".lock"
with open(lock_path, "x", encoding="utf-8"):
pass
text = read_with_retry(path, attempts, wait=wait)
os.remove(lock_path)
return text
test_main.py
import os
from main import ReadFailedError, read_locked
def clean(*names):
for name in names:
if os.path.exists(name):
os.remove(name)
def test_success_removes_lock():
"""After a good read the lock file is gone"""
clean("a.txt.lock")
with open("a.txt", "w", encoding="utf-8") as f:
f.write("A")
got = read_locked("a.txt")
assert got == "A", f"read_locked returned {got!r}"
assert not os.path.exists("a.txt.lock"), "a.txt.lock is still there after a good read"
def test_failure_removes_lock():
"""After a failed read the lock file is gone too"""
clean("b.txt", "b.txt.lock")
try:
read_locked("b.txt", attempts=2)
except ReadFailedError:
pass
else:
assert False, "read_locked returned normally for a missing file"
assert not os.path.exists("b.txt.lock"), "b.txt.lock is still there after a failed read; remove it in finally"
def test_existing_lock_is_kept():
"""If the lock exists, read_locked fails and leaves it alone"""
with open("c.txt.lock", "w", encoding="utf-8") as f:
f.write("held")
try:
read_locked("c.txt")
except FileExistsError:
pass
else:
assert False, "read_locked did not raise FileExistsError for a held lock"
assert os.path.exists("c.txt.lock"), "read_locked removed a lock it did not create"
clean("c.txt.lock")
def test_lock_already_gone():
"""A lock removed by someone else during the read is not an error"""
clean("d.txt", "d.txt.lock")
def wait(n):
clean("d.txt.lock")
with open("d.txt", "w", encoding="utf-8") as f:
f.write("D")
got = read_locked("d.txt", wait=wait)
assert got == "D", f"read_locked returned {got!r}"
On macOS and Linux, type python3 wherever these commands say python, as in the first lesson.
Run the program:
python main.pyRun the checks (needs learnrun.py in the same folder):
python learnrun.py testCommon mistakes
Cleanup after yield without finally
from contextlib import contextmanager
@contextmanager
def lock():
print("lock")
yield
print("unlock")
with lock():
raise OSError("disk full")
What Python prints
OSError: disk fullWhy, and the fix
The program prints lock but never unlock: the OSError comes back out of the yield and skips the line after it. Put the yield in try: and print("unlock") in finally:, and the cleanup runs however the block ends.
Forgetting @contextmanager
def lock():
print("lock")
yield
print("unlock")
with lock():
print("working")
What Python prints
TypeError: 'generator' object does not support the context manager protocol (missed __exit__ method)Why, and the fix
Without the decorator, lock() returns a plain generator, which with cannot use. Add from contextlib import contextmanager and put @contextmanager on the line above def lock():.
Retrying inside the context manager
from contextlib import contextmanager
@contextmanager
def retrying(attempts):
for n in range(attempts):
try:
yield
return
except OSError:
print("attempt", n + 1, "failed")
with retrying(3):
open("missing.txt")
What Python prints
RuntimeError: generator didn't stop after throw()Why, and the fix
A with block runs once, so its generator must yield exactly once. Yielding again after the first failure is an error. Put the retry loop in a function, as read_with_retry does, and use a context manager for one attempt.
Python in the browser: Pyodide 314.0.7, MPL-2.0. Licence and source
Exit ticket
5 questions, no hints. Score 80% or more to complete the lesson.
Finish every activity above to unlock the exit ticket.