How do I run asynchronous function in blocking style? I am not allowed to modify signature of my mock function
f1() and can’t easy switch to async def and so can’t use await expression.
async def cc2(): await asyncio.sleep(1) await asyncio.to_thread(print, "qwerty") def f1(): t = asyncio.create_task(cc2()) # wait until the task finished before returning async def main(): f1() # typical call from many places of a program asyncio.run(main())
asyncio.get_running_loop().run_until_complete(t), but the hack does not work and I get the next error.
Traceback (most recent call last): File "<stdin>", line 1, in <module> File "/usr/lib/python3.9/asyncio/runners.py", line 44, in run return loop.run_until_complete(main) File "/usr/lib/python3.9/asyncio/base_events.py", line 642, in run_until_complete return future.result() File "<stdin>", line 2, in main File "<stdin>", line 3, in f1 File "/usr/lib/python3.9/asyncio/base_events.py", line 618, in run_until_complete self._check_running() File "/usr/lib/python3.9/asyncio/base_events.py", line 578, in _check_running raise RuntimeError('This event loop is already running') RuntimeError: This event loop is already running
How do I run asynchronous function in blocking style?
If you are in sync code, you can call it with
asyncio.run(async_function()). If you are in async code, you can
await async_function() or
asyncio.create_task(async_function()), with the latter scheduling it to run in the background. You are not allowed to use
run_until_complete, even with a newly created event loop object) inside async code because it blocks and could halt the outer event loop.
But if you need it for testing purposes, you can always do something like:
# XXX dangerous - could block the current event loop with concurrent.futures.ThreadPoolExecutor() as pool: pool.submit(asyncio.run, async_function()).result()