从Python中的同步函数调用异步代码

Question

这个问题可能并不复杂，但我是异步库的新手，我只是想不通。

假设我有一个同步函数，我想从中调用异步函数，而不将同步函数转换为异步函数。 同步函数需要等待异步函数完成并返回它提供的值。如果它同时阻止执行也没关系。 所有这些都是必要的，因为异步函数来自库。

如何检索异步函数计算的结果？

• 已经有一个正在运行的事件循环，并且我有一个对它的引用。
• 它们在同一个线程上运行，所以我不需要另一个线程。
• 我无法使用 run_until_complete() 因为它在某种程度上与已经运行的事件循环发生冲突。

import asyncio

class Sample:

    def __init__(self):
        self.state = self.compute_sync()  # I want to call async from this constructor

    async def io_function(self):
        print("started")
        x = await asyncio.sleep(1)  # using an async library here
        print("ended")
        return x.result()

    def compute_sync(self):
        io_result = self.io_function()  # I want the result of this without using await
        return 'local_data' + io_result

async def main():
    await asyncio.sleep(1)    # other IO stuff
    s = Sample()
    print(s)

asyncio.run(main())

（上面的代码非常简化，也许它没有反映真实的用例，但注释可能会帮助我在这里需要做的事情）

• 我正在使用最新版本的 Python（当前为 3.13）寻找答案。
• 如果可能的话，我不想使用外部异步库。

Answer 1

你的评论说“我想要这个结果而不使用等待”。最简单的部分是创建一个带有所需结果的任务（任务的结果是协程的返回值）

task = asyncio.create_task(coro())` # create_task is a regular function.

真正的问题是如何以及何时获得结果。 （异步函数的）结果需要一段时间才能获得。 如果没有await

，你就无法等待

。尝试在同步函数中等待会使程序停止。

选项 1：以一定的时间间隔进行轮询。用

task.done()

检查，用

task.result()

检索结果。

选项 2：通过回调完成计算后立即采取行动：使用

task.add_done_callback()

。

---

但是，我首先考虑将有问题的常规/同步函数转换为异步版本。如果这种更改对于特定用例是可能的并且也很容易进行：

• 将函数定义从

  def

  更改为

  async def

• 始终用

  await

来称呼它

那么可以调用哪些函数类型（同步与异步）就没有限制。