在 FastAPI 中使用 `async def` 与 `def` 并测试阻塞调用

Question

tl;dr

1. 以下哪个选项是fastapi 中正确的工作流程？
2. 如何以编程方式测试呼叫是否真正阻塞（除了从浏览器手动）？ uvicorn 或 fastapi 是否有压力测试扩展？

我在fastapi 服务器中有许多端点（目前使用uvicorn），它们对常规同步 Python 代码的调用有很长时间的阻塞。尽管有文档 (https://fastapi.tiangolo.com/async/)，但我仍然不清楚我是否应该专门使用 def、async def 或混合用于我的功能。

据我所知，我有三个选择，假设：

def some_long_running_sync_function():
  ...

选项 1 - 始终将def 仅用于端点

@app.get("route/to/endpoint")
def endpoint_1:
  some_long_running_sync_function()


@app.post("route/to/another/endpoint")
def endpoint_2:
  ...

选项 2 - 始终使用 async def 并在执行程序中运行阻塞同步代码

import asyncio


@app.get("route/to/endpoint")
async def endpoint_1:
  loop = asyncio.get_event_loop()
  await loop.run_in_executor(None, some_long_running_sync_function)


@app.post("route/to/another/endpoint")
async def endpoint_2:
  ...

选项 3 - 根据底层调用混合搭配 def 和 async def

import asyncio


@app.get("route/to/endpoint")
def endpoint_1:
  # endpoint is calling to sync code that cannot be awaited
  some_long_running_sync_function()


@app.post("route/to/another/endpoint")
async def endpoint_2:
  # this code can be awaited so I can use async
  ...

Answer 1

由于没有人捡起这个，我给了我的两分钱。所以这仅源于我的个人经验：

选项 1

这完全违背了使用异步框架的目的。所以，可能，但没用。

选项 2 和 选项 3

对我来说，两者兼而有之。该框架明确地支持同步和异步端点的混合。通常我会选择：没有await in the function/CPU-bound task -> def, IO-bound/very short -> async.

陷阱

当您开始使用异步编程时，很容易陷入现在您不必再关心线程安全的谬论。但是一旦你开始在线程池中运行东西，这就不是真的了。所以请记住，FastAPI 在线程池中运行您的 def 端点，您有责任使它们成为线程安全的。

这也意味着您需要考虑在def 端点中您可以使用事件循环做什么和不可以做什么。由于您的事件循环在主线程中运行，asyncio.get_running_loop() 将不起作用。这就是为什么我有时会定义async def 端点，即使没有 IO，也能够从同一个线程访问事件循环。但是当然你必须保持你的代码简短。

附带说明：FastAPI 还公开了 starlettes backgroundtasks，它可以用作从端点创建任务的依赖项。

在我写这篇文章的时候，这似乎很固执，这可能是你直到现在还没有得到很多反馈的原因。所以，如果你不同意任何事情，请随意击落我:-)