Skip to content

← Everyday Standard Library Tools step 34 of 55

Medium Primitives

Retry a Flaky Call and Preset Arguments with partial

Network calls sometimes fail once and work on the next try. Write a retry wrapper that keeps the wrapped function's types, then add two preset methods to a Notifier class. Your solve puts both into one report.

waits: list[float] = []
retrier = make_retrier(flaky, attempts=3, delay=0.25, sleep=waits.append)
retrier(10)  # flaky fails twice, then the retrier returns 10
waits        # [0.25, 0.5]

The retry wrapper

def make_retrier(
    fn: Callable[P, R], *, attempts: int, delay: float,
    sleep: Callable[[float], None],
) -> Callable[P, R]: ...
  • The attempts argument is the total number of calls allowed, including the first. If it is below 1, raise ValueError from make_retrier itself, before calling fn or sleep.
  • On each attempt, pass every positional and keyword argument on to fn. Return its result as soon as a call succeeds.
  • Retry only on ValueError. Any other exception from fn, and any exception from sleep, goes straight to the caller.
  • After j failures, if another attempt remains, call sleep(delay * j). There is no wait before the first call or after the last failure, so three failed attempts wait delay and then 2 * delay.
  • If every attempt fails, raise the last ValueError.
  • Call sleep, never time.sleep. The delay is a finite float, 0 or more.
  • Type it with ParamSpec and TypeVar, so the result keeps the parameter and return types of fn. Copying the name with wraps is not required.

The notifier

  • shout(text: str, *, mark: str = "!") -> str returns text.upper() + mark.
  • Notifier(channel) has send(body: str, *, level: str = "info") -> str, which returns f"{self.channel}/{level}: {body}".
  • alert is a partialmethod of send with level="alert".
  • banner is a staticmethod around a partial of shout with mark="!!!". It must work through the class and through an instance.

The report

Write solve(fail_times: int, attempts: int, boost: int | None) -> dict[str, object]. Inside it, define flaky(base: int, *, boost: int = 1) -> int. It counts its calls, raises ValueError on the first fail_times calls, and then returns base * boost. If fail_times is 0 or less, it never fails.

Wrap flaky with attempts, a delay of 0.25, and waits.append as sleep. Call the retrier with base 10. Leave out boost when the input is None, and pass it otherwise, even when it is 0 or negative.

Key Value
outcome The result, or "exhausted" if every call failed
calls How many times flaky ran
waits The requested delays, as floats
alert Notifier("ops").alert("disk full"), which is "ops/alert: disk full"
banner Notifier("ops").banner("ship it"), which is "SHIP IT!!!"
partial_pickles Whether partial(str.upper, "revived") survives a pickle round trip and returns "REVIVED" when called
lambda_pickles Whether pickle can save a local lambda, which is False

For each pickle check, catch the exception and record False. If make_retrier rejects attempts, return this report and run nothing else:

{"outcome": "ValueError", "calls": 0, "waits": [], "alert": "", "banner": "",
 "partial_pickles": False, "lambda_pickles": False}

Your code must pass mypy --strict.