如何理解python大型项目后台异步，很多新手对此不是很清楚，为了帮助大家解决这个难题，下面小编将为大家详细讲解，有这方面需求的人可以来学习下，希望你能有所收获。
成都创新互联公司是一家专业提供金坛企业网站建设,专注与成都网站制作、成都做网站、成都h5网站建设、小程序制作等业务。10年已为金坛众多企业、政府机构等服务。创新互联专业网站设计公司优惠进行中。
1
BackgroundTasks 使用场景
有时候我们需要在 request 执行之后继续一些操作，但终端并不需要等待这些操作完成才能收到 response 。我列举一些场景大家看一下：
1.在自动出票完成后需要向各 ota 平台自动发送行程单信息
2.在执行完购票后需要向各户发送邮件通知购票成功信息
3.收到客户端的文件之后对文件进行二次处理
4....
5....
这些操作都需要一定的处理时间，但与返回给终端的 response 并无直接关系这个时候就可以通过定义后台任务 BackgroundTasks 来实现这个功能。
2
BackgroundTasks 实战
2.1
添加参数
首先我们需要导入 BackgroundTasks，并在路径操作函数中添加 BackgroundTasks 类型的参数。
# -*- encoding: utf-8 -*-
from fastapi import BackgroundTasks, FastAPI
app = FastAPI()
@app.post("/send_info")
async def send_notification(email: str, background_tasks: BackgroundTasks):
"""
发送提醒任务
"""
pass2.2
任务函数
任务函数是指：在需要创建一个在后台任务中实际执行的函数。任务函数是一个标准函数。这个函数可以是 async def 或者 普通 def 的函数。
这里创建一个把指定内容写入文件的函数。
# -*- encoding: utf-8 -*-
def write_notification(email: str, message=""):
"""
写入文件操作模拟任务
"""
with open("log_test.txt", mode="w") as email_file:
content = "notification for {}: {}".format(email,message)
email_file.write(content)2.3
添加后台任务
最后需要把任务函数添加到后台任务中
# -*- encoding: utf-8 -*-
import time
from fastapi import BackgroundTasks, FastAPI
app = FastAPI()
def write_notification(email: str, message=""):
"""
写入文件操作模拟任务
"""
time.sleep(5)
with open("log_test.txt", mode="w") as email_file:
content = "notification for {}: {}".format(email,message)
email_file.write(content)
print("write end")
time.sleep(2)
@app.post("/send_info")
async def send_notification(email: str, background_tasks: BackgroundTasks):
background_tasks.add_task(write_notification, email, message="some notification")
return {"message": "now: %s Notification sent in the background">2.4
add_task 解释
.add_task()接收以下参数信息：1.在后台运行的任务函数(例如 write_notification)2.任意序列的参数信息(例如 email)3.任意键值对的参数信息(例如 message="some notification")4.我们故意在 write_notification 方法中加入等待时间 来验证对于客户端的返回
2.5
依赖注入
后台任务可以与依赖注入系统一起使用，可以在不同层级的依赖项中声明 BackgroundTasks 参数
# -*- encoding: utf-8 -*-
from typing import Optional
from fastapi import BackgroundTasks, Depends, FastAPI
app = FastAPI()
def write_log(message: str):
with open("log.txt", mode="a") as log:
log.write(message)
def get_query(background_tasks: BackgroundTasks, q: Optional[str] = None):
if q:
message = f"found query: {q}\n" background_tasks.add_task(write_log, message)
return q
@app.post("/send_info")
async def send_notification(email: str, background_tasks: BackgroundTasks, q: str = Depends(get_query)):
message = f"message to {email}\n" background_tasks.add_task(write_log, message)
return {"message": "Message sent"}1.若需要执行大量的后台计算而不必一定要在同一进程中运行，例如：它不需要共享内存，变量等，则可使用其他更大的工具，例如:celery、MQ 系列 都是可以选择的但这些往往需要更复杂的配置，例如:RabbitMQ、redis 之类的消息作业队列管理器，但是它们允许在多个进程（尤其是多个服务器）中运行后台任务。
2.若需要从同一 FastAPI 应用访问变量和对象，或者需要执行一些小的后台任务 例如:发送电子邮件、短信消息等，则只需使用即可 BackgroundTasks。
看完上述内容是否对您有帮助呢？如果还想对相关知识有进一步的了解或阅读更多相关文章，请关注创新互联行业资讯频道，感谢您对创新互联的支持。
