介绍

FastAPI是一个用于用 Python 编写 API 的领先 Web 框架，拥有卓越的速度、用户友好的属性和出色的异步功能。其中，一个突出的功能是FastAPI的BackgroundTasks——一种创新工具，旨在管理长时间运行、耗时的任务，而不会抑制主要应用程序操作。

在这篇博文中，我们将深入研究 FastAPI 的后台任务，阐明它们如何显着提高应用程序性能和响应能力。我们将通过一个简单的 FastAPI 应用程序的逐步增强来记录这一旅程，演示将长时间运行的任务精心卸载到后台时的转换。

创建简单的 FastAPI 应用程序

在我们给定的Python代码片段中，我们展示了一个基本的FastAPI应用程序，它面临着一个关键挑战——处理由very_long_task ()模拟的长时间运行的操作。此函数引入了延迟，演示了耗时的任务如何阻塞应用程序。

import time   
import uvicorn   
from fastapi import FastAPI   
  
app = FastAPI()   
  
def  very\_long\_task ():   
   print ( "开始很长的任务" )   
   time.sleep( 5 )   
   print ( "结束很长的任务" )   
  
@app.post( "/long\_task " )   
def  long\_task\_endpoint ():   
   print ( "进入端点 " )   
   very\_long\_task()   
   print ( "退出端点 " )   
   return { "status" :"very\_long\_task结束" }   
  
if \_\_name\_\_ == "\_\_main\_\_" :   
   uvicorn.run( "app:app" , host= "0.0.0.0" , port= 8000 , reload= True )

调用long_task端点时，您将观察到以下日志序列：

进入端点
开始很长的任务
结束很长的任务
退出端点

此实现中的障碍是与我们的应用程序的连接在执行期间挂起very_long_task。保持连接打开这么长时间通常被认为是不好的做法。如果在此期间连接中断，则长任务将不会执行，从而导致潜在的数据不一致或其他不良影响。

现在我们将探讨如何使用 FastAPI 的后台任务来处理这种情况。请继续关注我们深入研究 FastAPI 中的异步编程世界，以提高应用程序的响应能力和可靠性。

解决阻塞问题

FastAPI 中后台任务的使用显着改变了我们的应用程序处理耗时操作的方式，有效提高了性能和响应能力。这一修改将我们的阻塞操作变成了非阻塞操作，允许应用程序在后台运行冗长的操作时继续处理其他请求。

import time   
from collections import OrderedDict   
  
import uvicorn   
from fastapi import BackgroundTasks, FastAPI, HTTPException   
  
app = FastAPI()   
  
def  very\_long\_task ():   
   print ( "开始很长的任务" )   
   time.sleep( 5 )   
   print ( "结束很长的任务" )   
  
@app.post( "/long\_task" )   
def  long\_task\_endpoint ( background\_tasks:BackgroundTasks ):   
   print ( "进入端点" )  
   background\_tasks.add\_task(very\_long\_task)   
   print ( "退出端点 " )   
   return { "status" : "非常长的任务正在运行" }   
  
if \_\_name\_\_ == "\_\_main\_\_" :   
   uvicorn.run( "app:app" , host= "0.0. 0.0"，端口= 8000，重新加载= True）

以下是对所做修改的更深入的了解：

我们的函数现在接受一个类型为 的long_task_endpoint()附加参数。该参数允许 FastAPI 在返回响应后安排在后台执行的函数。background_tasks`BackgroundTasks`

在内部long_task_endpoint()，我们使用对象add_task()上的方法background_tasks将函数添加very_long_task()到后台任务列表中。我们的函数现在不再直接执行very_long_task()并等待它完成，而是简单地将其添加到后台任务中并继续，使其立即返回响应。

我们还更新了响应：我们现在返回一个指示长任务正在运行的响应，而不是通知客户端任务已结束。这是因为任务现在计划在后台运行，并且响应是在任务完成之前发送的。

从本质上讲，这个更新的 FastAPI 应用程序显示了后台任务在管理长时间运行的操作方面的强大潜力。通过这种方法，我们的应用程序变得更加高效和响应迅速，能够处理多个请求，而不会受到耗时任务的阻碍。

奖励：检查任务的状态

在最新的迭代中，我们引入了一种新的 GET 方法。该端点/status现在使用户能够检查长时间运行的任务的状态，从而深入了解应用程序中特定操作的进度。

import time   
import uuid   
import uvicorn   
from fastapi import BackgroundTasks, FastAPI, HTTPException   
  
app = FastAPI()   
  
status\_dict = {}   
  
def  very\_long\_task ( task\_id ):   
   print ( "开始很长的任务" )   
   try :   
       time.sleep( 5 )   
       status\_dict\[ str ( task\_id)\] = “已完成”，例外  
   为e :   
       status\_dict\[ str (task\_id)\] = “失败”   
   print ("结束很长的任务" )   
  
@app.get( "/status" )   
def  get\_status ( task\_id: str ):  
   如果task\_id不在 status\_dict中:   
       raise HTTPException(status\_code= 404 ,Detail= "Item not found" )   
   return status\_dict\[task\_id \]   
  
@app.post( "/long\_task" )   
def  long\_task\_endpoint ( background\_tasks: BackgroundTasks ):   
   print ( "正在进入端点 " )   
   task\_id = uuid.uuid1()   
   status\_dict\[ str(task\_id)\] = "正在运行"  
    background\_tasks.add\_task(very\_long\_task, task\_id)   
   print ( "退出端点 " )   
   return task\_id   
  
if \_\_name\_\_ == "\_\_main\_\_" :   
   uvicorn.run( "app:app" , host= "0.0.0.0"，端口= 8000，重新加载= True）

本次迭代的主要更新包括以下增强功能：

• 我们引入了一个端点 ，/status它接受task_id我们想要用来检查所需任务的状态的 。如果未找到任务，则会HTTPException引发错误，从而导致404错误。
• 为了启用此功能，我们创建了一个status_dict字典来跟踪所有任务的状态。
• 在端点内，生成long_taska ，然后添加到初始状态为“正在运行”的 。这也被传递到主任务函数，在其中执行主操作。task_id`status_dicttask_idvery_long_task`
• 成功执行任务后，状态status_dict将更新为“已完成”。如果执行期间发生故障，状态将设置为“失败”。

请记住，在现实生活中的用例中，您不应将状态存储在全局变量中。这种方法的问题是，如果重新启动应用程序，您将丢失所有状态。此外，您无法控制 ; 的大小status_dict。在严重情况下，它可能会溢出主机上的内存。更好的方法是使用数据库来存储状态。另一种选择是利用队列系统（例如 Celery）来更有效地管理这些任务。

结论

在本文中，我们探讨了 FastAPI 实现中的一个更高级的主题：后台任务。这一功能使我们能够通过将长时间的阻塞操作转换为非阻塞操作来优化应用程序的性能。此外，我们还实施了一个系统来检查正在进行的任务的状态，确保应用程序响应更快且用户友好。这些技术是增强基于 FastAPI 的项目的效率和功能的强大方法。