Write a Sync Wrapper for Async Code

Libraries often offer a sync version of an async function. Write run_sync, a general helper that takes any coroutine and runs it from regular code.

async def double(n: int) -> int:
    await asyncio.sleep(0)
    return n * 2

run_sync(double(21))  # 42 from a plain script, RuntimeError from async code

What to write

def run_sync[T](coro: Coroutine[Any, Any, T]) -> T:

coro is a fresh coroutine that nobody has awaited yet. The [T] after the name makes the function generic. T stands for the coroutine's result type, so run_sync(double(21)) has type int.

• Check for a running loop with asyncio.get_running_loop(). It returns the loop running in the calling thread, or raises RuntimeError if there is none. Put only that call inside the try, so that a RuntimeError raised by the coroutine itself still reaches the caller.
• When get_running_loop() raises, run coro with asyncio.run and return its result. Each call gets its own new loop, and asyncio.run closes it at the end, whether the coroutine succeeded or failed.
• If the coroutine raises, let the exception propagate with its original type, message and traceback.
• When get_running_loop() returns a loop, call coro.close() and then raise RuntimeError(NESTED_MESSAGE), using the constant from the starter. Close the coroutine without starting a loop or running it. Closing it also silences the warning Python would print later about an unawaited coroutine.

Edge cases

• Only the calling thread counts. In the other_thread scenario, async code calls run_sync through asyncio.to_thread, which runs it on a worker thread. That thread starts without a loop, so the call succeeds.
• Return a result of 0 or a negative number as it is.

What the driver reports

Keep solve and the helper coroutines as they are. The scenario input picks one of five runs:

• value makes one call, and raises calls a coroutine that raises ValueError.
• nested calls run_sync from inside a running loop.
• sequential makes repeat calls in a row. Only this scenario uses repeat.
• other_thread calls from a worker thread.

Each report has these keys:

• outcome is "ok", or the name of the exception the call raised. message is the exception's text.
• value is the last successful result. It starts at -1 and keeps that value when a call fails, or when sequential gets a repeat of zero or less.
• traceback_ok is True when the traceback still shows _boom, the coroutine that raised.
• loops_seen counts the loops the helper coroutines ran on. A refused call runs nothing, so it records zero.
• distinct_loops is True when every recorded loop is a different object, and all_closed is True when every recorded loop is closed afterwards.