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

Question

我正在使用FastAPI和uvloop以高效的方式为RESTAPI提供服务

我有很多异步代码可以调用远程资源，如数据库、存储器等，这些函数如下所示：

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是一个阻塞调用，在异步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绑定操作，则将其执行委托给单独的进程（最简单的方法是使用ProcessPoolExecutor或任何任务队列执行run_in_executor）