Build a Typed Async Wrapper for Blocking Calls

Build two helpers that let async code call blocking functions without blocking the event loop. asyncify turns one blocking function into an async one. gather_blocking runs a batch of blocking functions on a thread pool of its own.

joined = asyncify(join)   # join(n: int, *, sep: str = ":") -> str
await joined(7, sep="-")  # "7-7", computed in a worker thread
joined.__name__           # "join"

await gather_blocking([lambda: 1, lambda: 2, lambda: 3], limit=2)  # [1, 2, 3]

What to write

def asyncify[**P, T](fn: Callable[P, T]) -> Callable[P, Coroutine[Any, Any, T]]

async def gather_blocking[T](fns: Sequence[Callable[[], T]], *, limit: int) -> list[T]

asyncify(fn) returns an async function that follows these rules:

• It runs fn in a worker thread with asyncio.to_thread and returns the result.
• It passes positional and keyword arguments through unchanged.
• It keeps the __name__ and __doc__ of fn. Use functools.wraps.
• An exception raised by fn reaches the caller with its own type and message.
• Its type uses the ParamSpec P, as in the article, not Callable[..., Any].

gather_blocking(fns, limit=...) follows these rules:

• It creates its own ThreadPoolExecutor(max_workers=limit, thread_name_prefix="gb").
• It runs each callable on that pool with loop.run_in_executor. Await each one in its own task inside one TaskGroup, so a failing job raises an exception group.
• It returns the results in the same order as fns.
• It always shuts the pool down before it returns or raises. Call pool.shutdown(wait=True) through asyncio.to_thread, so the loop keeps running while the pool's threads finish.
• If it is cancelled while it waits for the shutdown, it keeps waiting until the shutdown is done, then re-raises the CancelledError.

Edge cases

• An empty fns returns [].
• A limit below 1 raises ValueError, even when fns is empty. ThreadPoolExecutor raises it for you.
• Cancelling a task does not stop a job that is already running in its thread. The driver's jobs always finish, so every shutdown completes.

What the driver reports

solve runs one scenario. Meanwhile a heartbeat task ticks five times, then sets the signal that the blocking functions wait for.

• ticks: how many times the heartbeat ticked, which is 5. If your code blocks the loop, the heartbeat cannot run. The blocking function then gives up after 3 seconds, and value comes back as "stalled".
• value: the result of the wrapped join call.
• name and doc: the __name__ and __doc__ of your wrapper.
• outcome and message: "ok" and "", or the type name and message of the exception the call raised.
• values: the list gather_blocking returned.
• leftover_threads: how many threads whose names start with gb are still alive after your helper returns. It must be 0.