I am using a third party module to retrieve data from an API. I simply would like to asynchronously await the module to return the data which occasionally takes several seconds and freezes up my app. However, when I try to await a call to that module I receive the TypeError:
TypeError: object dict can't be used in 'await' expression
import thirdPartyAPIwrapper
async def getData():
retrienveData = await thirdPartyAPIWrapper.data()
return await retrieveData
def main():
loop = asncio.get_event_loop()
data = loop.run_until_complete(getData())
loop.close
return data
Why can I not await a type('dict')? Is there a way around this? If async/await with asyncio will not work with a third party module that doesn't return a coroutine then what are my other options?
Only asynchronous (defined with async def
) functions can be awaited. Whole idea is that such functions are written special way what makes possible to run (await
) them without blocking event loop.
If you want to get result from common (defined with def
) function that takes some considerable time to be executed you have these options:
Usually you want to choose second option.
Here's example of how to do it:
import asyncio
import time
from concurrent.futures import ThreadPoolExecutor
_executor = ThreadPoolExecutor(1)
def sync_blocking():
time.sleep(2)
async def hello_world():
# run blocking function in another thread,
# and wait for it's result:
await loop.run_in_executor(_executor, sync_blocking)
loop = asyncio.get_event_loop()
loop.run_until_complete(hello_world())
loop.close()
Please, read this answer about how asyncio works. I think it'll help you much.
As thirdPartyAPIWrapper.data()
is a normal sync function you should call it in another thread.
There is a helper function for that in a asgiref
library.
Assume we've got a blocking function with an argument:
import asyncio
import time
from asgiref.sync import sync_to_async
def blocking_function(seconds: int) -> str:
time.sleep(seconds)
return f"Finished in {seconds} seconds"
async def main():
seconds_to_sleep = 5
function_message = await sync_to_async(blocking_function)(seconds_to_sleep)
print(function_message)
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()
There is also an async_to_sync
helper function in that library.
You do not need to await which function is running asynchronously
async def getData():
retrienveData = thirdPartyAPIWrapper.data()
return retrieveData
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With