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

Question

tl;dr

1. 以下哪个选项是正确的工作流程fastapi？
2. 如何以编程方式测试调用是否真正被阻止（而不是从浏览器手动测试）？是否有对or 的压力测试扩展？uvicornfastapi

我在服务器中有许多端点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

由于没有人注意到这一点，我\xe2\x80\x98m 给出了我的两分钱。所以这仅源于我的个人经历：

\n

选项1

\n

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

\n

选项 2和选项 3

\n

对我来说，它\xe2\x80\x98是两者的混合体。该框架明确支持同步和异步端点的混合。一般来说，我会选择：await函数/CPU 密集型任务中否 -> def，IO 密集型/非常短 -> async。

\n

陷阱

\n

当您开始使用异步编程时，很容易陷入这样的谬论：现在您不必再关心线程安全了。但是一旦你开始在线程池中运行东西，这就不再是\xe2\x80\x98了。因此请记住，FastAPI 在线程池中运行您的def端点，您有责任确保它们的线程安全。

\n

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

\n

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

\n

当我\xe2\x80\x98m 写这篇文章时，这似乎很固执己见，这可能是你到目前为止\xe2\x80\x98t 没有得到很多反馈的原因。因此，如果您不同意任何内容，请随时拒绝我:-)

\n