Can I use asyncio.run() to run coroutines multiple times?

This actually is an interesting and very important question.

As a documentation of asyncio (python3.9) says:

This function always creates a new event loop and closes it at the end. It should be used as a main entry point for asyncio programs, and should ideally only be called once.

It does not prohibit calling it multiple times. And moreover, an old way of calling coroutines from synchronous code, which was:

loop = asyncio.get_event_loop() loop.run_until_complete(coroutine) 

Is now deprecated because of get_event_loop() method, which documentation says:

Consider also using the asyncio.run() function instead of using lower level functions to manually create and close an event loop.

Deprecated since version 3.10: Deprecation warning is emitted if there is no running event loop. In future Python releases, this function will be an alias of get_running_loop().

So in future releases it will not spawn new event loop if already running one is not present! Docs are proposing usage of asyncio.run() if You want to automatically spawn new loop if there is no new one.

There is a good reason for such decision. Even if You have an event loop and You will successfully use it to execute coroutines, there is few more things You must remember to do:

• closing an event loop
• consuming unconsumed generators (most important in case of failed coroutines)
• ...probably more, which I do not even attempt to refer here

What is exactly needed to be done to properly finalize event loop You can read in this source code.

Managing an event loop manually (if there is no running one) is a subtle procedure, and it is better to not doing that, unless one know what he is doing.

So Yes, I think that proper way of runing async function from synchronous code is calling asyncio.run(). But it is only suitable from a fully synchronous application. If there is already running event loop, it will probably fail (not tested). In such case, just await it or use get_runing_loop().run_untilcomplete(coro).

And for such synchronous apps, using asyncio.run() it is safe way and actually the only safe way of doing this, and it can be invoked multiple times.

The reason docs says that You should call it only once is that usually there is one single entrypoint to whole asynchronous application. It simplifies things and actually improves performance, because setting thins up for an event loop also takes some time. But if there is no single loop available in Your application, You should use multiple calls to asyncio.run() to run coroutines multiple times.

Is there is any performance gain?

Beside discussing multiple calls to asyncio.run(), I want to address one more concern. In comments, @jwal says:

asyncio is not parallel processing. Says so in the docs. [...] If you want parallel, run in a separate processes on a computer with a separate CPU core, not a separate thread, not a separate event loop.

Suggesting that asyncio is not suitable for parallel processing, which can be misunderstood and misleading to a conclusion, that it will not result in a performance gain, which is not always true. Moreover it is usually false!

So, any time You can delegate a job to an external process (not only a python process, it can be a database worker process, http call, ideally any TCP socket call) You can utilize a performance gain using asyncio. In huge majority of cases, when You are using a library which exposes async interface, the author of that library made an effort to eventually await for a result from a network/socket/process call. While response from such socket is not ready, event loop is completely free to do any other tasks. If loop has more than one such tasks, it will gain a performance.

A canonical example of such case is making a calls to a HTTP endpoints. At some point, there will be a network call, so python thread is free to do other work while awaiting for a data to appear on a TCP socket buffer. I have an example!

The example uses httpx library to compare performance of doing multiple calls to a OpenWeatherMap API. There are two functions:

• get_weather_async()
• get_weather_sync()

The first one does 8 request to an http API, but schedules those request to run cooperatively (not concurrently!) on an event loop using asyncio.gather().

The second one performs 8 synchronous request in sequence.

To call the asynchronous function, I am actually using asyncio.run() method. And moreover, I am using timeit module to perform such call to asyncio.run() 4 times. So in a single python application, asyncio.run() was called 4 times, just to challenge my previous considerations.

from time import time import httpx import asyncio import timeit from random import uniform class AsyncWeatherApi: def __init__( self, base_url: str = "https://api.openweathermap.org/data/2.5" ) -> None: self.client: httpx.AsyncClient = httpx.AsyncClient(base_url=base_url) async def weather(self, lat: float, lon: float, app_id: str) -> dict: response = await self.client.get( "/weather", params={ "lat": lat, "lon": lon, "appid": app_id, "units": "metric", }, ) response.raise_for_status() return response.json() class SyncWeatherApi: def __init__( self, base_url: str = "https://api.openweathermap.org/data/2.5" ) -> None: self.client: httpx.Client = httpx.Client(base_url=base_url) def weather(self, lat: float, lon: float, app_id: str) -> dict: response = self.client.get( "/weather", params={ "lat": lat, "lon": lon, "appid": app_id, "units": "metric", }, ) response.raise_for_status() return response.json() def get_random_locations() -> list[tuple[float, float]]: """generate 8 random locations in +/-europe""" return [(uniform(45.6, 52.3), uniform(-2.3, 29.4)) for _ in range(8)] async def get_weather_async(locations: list[tuple[float, float]]): api = AsyncWeatherApi() return await asyncio.gather( *[api.weather(lat, lon, api_key) for lat, lon in locations] ) def get_weather_sync(locations: list[tuple[float, float]]): api = SyncWeatherApi() return [api.weather(lat, lon, api_key) for lat, lon in locations] api_key = "secret" def time_async_job(repeat: int = 1): locations = get_random_locations() def run(): return asyncio.run(get_weather_async(locations)) duration = timeit.Timer(run).timeit(repeat) print( f"[ASYNC] In {duration}s: done {len(locations)} API calls, all" f" repeated {repeat} times" ) def time_sync_job(repeat: int = 1): locations = get_random_locations() def run(): return get_weather_sync(locations) duration = timeit.Timer(run).timeit(repeat) print( f"[SYNC] In {duration}s: done {len(locations)} API calls, all repeated" f" {repeat} times" ) if __name__ == "__main__": time_sync_job(4) time_async_job(4) 

At the end, a comparison of performance was printed. It says:

[SYNC] In 5.5580058859995916s: done 8 API calls, all repeated 4 times [ASYNC] In 2.865574334995472s: done 8 API calls, all repeated 4 times 

Those 4 repetitions was just to show that You can safely run a asyncio.run() multiple times. It had actualy destructive impact on measuring performance of asynchronous http calls, because all 32 request was actually run in four synchronous batches of 8 asynchronous tasks. Just to compare performance of one batch of 32 request:

[SYNC] In 4.373898585996358s: done 32 API calls, all repeated 1 times [ASYNC] In 1.5169846520002466s: done 32 API calls, all repeated 1 times 

So yes, it can, and usually will result in performance gain, if only proper async library is used (if library exposes an async API, it usually does it intentianally, knowing that there will be a network call somewhere).