In our project, a number of functions come in 2 flavors: asynchronous and synchronous. We write the asynchronous version and provide a runa() wrapper around the asyncio.run() function: async def read_sensor_async(): print("Reading sensor...") # Actual code return 9 # Simulate sensor reading def read_sensor(): return runa(read_sensor_async()) def runa(func): # Prepare steps... return asyncio.run(func) Things seem to be alright up to this point. However, when we run read_sensor() in a Jupyter notebook (which means inside an event loop). We get a RuntimeError for calling asyncio.run(). Reading the documentation, this is the expected behavior. So, to deal with this situation, we need to prepare runa to work both inside and outside an event loop. Here is our version 2: def runa(func): # Prepare steps... with contextlib.suppress(RuntimeError): return asyncio.run(func) # Gets here means we are inside an event loop return await func The problem with this version is since runa() is a synchronous, we cannot use await. What can we do to call func? Update A better question would be: How can we rewrite the function read_sensor() to work synchronously and asynchronously? Continue reading...
