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

问题描述

有很多库使用其自定义版本的Future。kafka和s3transfer只是两个例子：它们的所有类似未来的自定义类都将object作为超类。

毫不奇怪，您不能对此类对象直接调用asyncio.wrap_future()，也不能对它们使用await。

包装此类期货以便与Asyncio一起使用的正确方式是什么？

---

解决方案

如果未来类支持完成回调和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的线程调用时将失败。