且构网

分享程序员开发的那些事...
且构网 - 分享程序员编程开发的那些事

如何在Python中包装自定义的未来来与异步接口一起使用?

更新时间:2022-05-12 23:36:32

如果未来类支持完成回调和result方法等标准未来功能,则只需使用如下内容:

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

将该代码视为模板,您可以对其进行自定义以匹配您正在处理的未来的具体情况。

预期用途是从运行Asyncio事件循环的线程调用它:

value = await wrap_future(some_foreign_future)

如果从不同的线程调用它,请确保显式传递loop,因为asyncio.get_event_loop在从未注册到asyncio的线程调用时将失败。