Skip to content

← Precise Type Hints with Literals and Generics step 17 of 26

Hard Framework

Type the Callbacks of a Retry Helper

A network call sometimes fails, and trying again often works. Write retry, which calls an operation up to a set number of times. A callback, should_retry, looks at each failure and decides whether to try again. retry returns whatever the operation returns, and mypy keeps its type.

def fetch_count() -> int: ...

count = retry(fetch_count, is_transient, 3)  # count is an int

Which callbacks fit

Callable[[A], R] is the type of a function that takes an A and returns an R. retry takes should_retry: Callable[[BaseException], bool], so it may call the callback with any exception.

A callback fits when it accepts every argument that retry may pass. accept_anything(exc: object) fits, because every exception is an object. The starter's is_transient(exc: ValueError) does not fit. retry may pass it a KeyError, but its parameter type allows only ValueError.

This rule is contravariance, which goes the other way from Sequence. A function that accepts any Animal can stand in for one that accepts a Dog. So a callback may take a wider parameter type than the annotation names. Return types follow the usual direction. A function that returns a Dog can stand in for one that returns an Animal, so return types are covariant.

What to write

def retry[T](
    operation: Callable[[], T],
    should_retry: Callable[[BaseException], bool],
    attempts: int,
) -> T:
    ...

attempts is the most times retry calls operation, counting the first call. Retry at once, with no delay.

  • If attempts < 1, raise ValueError("attempts must be >= 1") before you call either callback.
  • Call operation. If it returns, return its value at once, even when the value is None, False or 0. Do not call should_retry on a success.
  • If operation raises an Exception, call should_retry with that exception object. Do this on the final attempt too.
  • If should_retry returns False, re-raise the exception with a bare raise, and make no more calls.
  • If it returns True and attempts remain, call operation again. If no attempts remain, re-raise the last exception from operation.
  • If should_retry itself raises, let that exception propagate. Do not retry it.
  • Catch only Exception. KeyboardInterrupt, SystemExit and any other BaseException outside Exception pass straight through, and should_retry never sees them.
  • Keep T in the return type, so an operation that returns int gives the caller an int.

Write two predicates:

  • is_transient(exc: BaseException) -> bool returns isinstance(exc, ValueError). Fix both the starter's annotation and its body.
  • accept_anything(exc: object) -> bool returns exc is not None, so it returns True for every exception that retry catches.

In solve, store the chosen predicate in a variable annotated Callable[[BaseException], bool]. Choose accept_anything when broad is True, and is_transient otherwise. Do not add cast or # type: ignore to that line, because that line is where mypy checks that both functions fit.

What solve returns

solve(outcomes, attempts, broad) builds an operation from the list outcomes. Its first call reads outcomes[0], its second call reads outcomes[1], and so on:

  • "ok" returns "value".
  • "ValueError" raises ValueError("transient").
  • Any other string raises KeyError("permanent").
  • Once the list runs out, every call returns "value".

Count every call, starting from zero on each solve call, and leave outcomes unchanged. Call retry with the operation and the chosen predicate. Return ("ok", result, calls) on success. If an Exception escapes, return ("failed", type(exc).__name__, calls). So an invalid attempts gives ("failed", "ValueError", 0).

solve(["ValueError", "ok"], 3, False)              # ("ok", "value", 2)
solve(["ValueError", "KeyError", "ok"], 5, False)  # ("failed", "KeyError", 2)
solve(["KeyError", "ok"], 3, True)                 # ("ok", "value", 2)
solve(["ValueError", "ok"], 1, False)              # ("failed", "ValueError", 1)
solve([], 2, False)                                # ("ok", "value", 1)
solve(["ok"], 0, True)                             # ("failed", "ValueError", 0)

In the second example, call 1 raises ValueError and retry tries again. Call 2 raises KeyError, which is_transient rejects, so retry stops there.

The typing specification lists the rules for which callables fit.