Skip to content

← Interfaces with Protocols and ABCs step 14 of 26

Hard Framework

Write Decorators That Keep the Signature with ParamSpec

You wrap a function in a decorator that retries it. If the decorator takes and returns Callable[..., Any], mypy stops checking calls to the wrapped function. flaky("oops") passes, though flaky needs a list and a string, and the result has the type Any.

In this exercise you write two decorators, retry and with_session, that keep the signature of the function they wrap.

Keep the parameters with ParamSpec

A ParamSpec is a type parameter that stands for a function's whole parameter list, so a decorator keeps the signature of the function it wraps. Write it with two stars, next to an ordinary type parameter for the return type:

def retry[**P, R](fn: Callable[P, R]) -> Callable[P, R]: ...

P is whatever parameters fn has, and R is its return type. The decorator returns a callable with the same parameters and return type. Now mypy reports both mistakes in flaky("oops").

Inside the decorator, annotate the wrapper with *args: P.args and **kwargs: P.kwargs, and give it the return type R. Use P.args and P.kwargs together, and only on *args and **kwargs. Also apply functools.wraps(fn) to the wrapper. It copies the name, docstring and other details of fn. ParamSpec does not.

Remove a leading parameter with Concatenate

with_session passes the first argument itself:

def with_session[**P, R](fn: Callable[Concatenate[Session, P], R]) -> Callable[P, R]: ...

Concatenate[Session, P] means a Session first, then the parameters in P. The decorated function takes only P. So callers of run_query pass sql and, if they want, the keyword-only upper, but never a session. mypy rejects run_query(123), and also run_query("select 1", True), which passes upper by position. Concatenate works only with positional parameters at the front of the list.

What to write

Keep ATTEMPTS = 3, and give both decorators the signatures above.

  • retry returns the result as soon as a call succeeds. It retries only on RuntimeError, and makes at most three calls. Any other exception propagates at once. If all three calls fail, raise RuntimeError(f"gave up after {ATTEMPTS} attempts") with raise ... from last, where last is the final error.
  • with_session creates a new Session() on every call and calls fn(Session(), *args, **kwargs).
  • Apply functools.wraps(fn) to both wrappers.

Then write solve(failures, labels, queries). failures and labels have the same length, and every count is zero or more.

  1. For each (fail, label) pair, in order, call flaky([fail + 1], label). Append its result, or the error message if retry gave up.
  2. Then, for each query in order, append run_query(query) and then run_query(query, upper=True).

flaky fails fail times before it succeeds. So counts of 0, 1 or 2 succeed within three attempts, and 3 or more give up. Each run_query result shows only the open and close events of its own call, because each call gets a new session. See PEP 612 for the full rules of ParamSpec and Concatenate.