There is a lot of libraries that use their custom version of Future. kafka and s3transfer are just two examples: all their custom future-like classes have object as the superclass.
Not surprisingly, you cannot directly call asyncio.wrap_future() on such objects and can't use await with them.
What is the proper way of wrapping such futures for use with asyncio?
If the future class supports standard future features such as done callbacks and the result method, just use something like this:
def wrap_future(f):
loop = asyncio.get_event_loop()
aio_future = loop.create_future()
def on_done(*_):
try:
result = f.result()
except Exception as e:
loop.call_soon_threadsafe(aio_future.set_exception, e)
else:
loop.call_soon_threadsafe(aio_future.set_result, result)
f.add_done_callback(on_done)
return aio_future
Consider that code a template which you can customize to match the specifics of the future you are dealing with.
Intended usage is to call it from the thread that runs the asyncio event loop:
value = await wrap_future(some_foreign_future)
If you are calling it from a different thread, be sure to pass loop explicitly because asyncio.get_event_loop will fail when invoked from a thread not registered with asyncio.
kafkaors3transfer-- perhaps you don't want to work with asyncio code. At least mixingkafkacalls inside an executor with regular async program gives expectedly slow result. The same is fors3transfer. What you need isaiokafkaandaiobotocoreusage.aiokafka0.4.0 in the first place, but unfortunately it's way too buggy -- e.g., it just hangs inposition()call or throws obscure exceptions. Just replacing it withkafkainstantly solved all our problems.call_soon_threadsafe(shown in the answer) orrun_coroutine_threadsafe.