Python：当已经有一个事件循环正在运行时，从同步方法调用异步代码

Question

我正在使用 FastAPI 和 uvloop 以高效的方式提供 REST API。

我有很多异步代码调用远程资源，例如数据库、存储等，这些函数看起来像这样：

async def _get_remote_resource(key: str) -> Resource:
    # do some async work
    return resource

我正在实现一个现有抽象基类的接口，我需要在同步方法中使用上面的异步函数。我做了类似的事情：

class Resource:
     def __str__(self):
         resource = asyncio.run_until_complete(_get_remote_resource(self.key))
         return f"{resource.pk}"

太棒了！现在我在 fastapi 中做了一个端点以使这项工作可以访问：

@app.get("")
async def get(key):
     return str(Resource(key))

问题是 FastAPI 已经获取并且事件循环正在运行，使用 uvloop，然后异步代码失败，因为循环已经在运行。

有什么方法可以从类中的同步方法调用异步方法吗？还是我必须重新考虑代码的结构？

Answer 1

运行时错误旨在防止您尝试执行的操作。 run_until_complete 是一个阻塞调用，在 async def 中使用它会停止外部事件循环。

直接的解决方法是通过实际的异步方法公开所需的功能，例如：

class Resource:
    def name(self):
        return loop.run_until_complete(self.name_async())

    async def name_async(self):
        resource = await _get_remote_resource(self.key)
        return f"{resource.pk}"

然后在 fastapi 中，您将以本机方式访问 API：

@app.get("")
async def get(key):
     return await Resource(key).name_async()

您也可以定义 __str__(self) 以返回 self.name()，但最好避免这样做，因为像 str() 这样基本的东西也应该可以从 asyncio 中调用（由于用于日志记录、调试等）。

Answer 2

我想补充@user4815162342 的答案。

FastAPI 是一个异步框架。我建议坚持几个原则：

• 不要以阻塞方式在同步函数中执行 IO 操作。异步准备这个资源，并且已经将准备好的数据传递给同步函数（这个原理可以称为同步代码的异步依赖）。
• 如果您仍需要在同步代码中执行阻塞 IO 操作，请在单独的线程中执行。并通过asyncio（def 端点、run_in_executor 与ThreadPoolExecutor 或def 后台任务）异步等待此结果。
• 如果您需要执行阻塞 CPU 密集型操作，则将其执行委托给单独的进程（最简单的方法是 run_in_executor 和 ProcessPoolExecutor 或任何任务队列）。