点击python编程从入门到实践，置顶 公众号重磅 python入门资料，第一时间送达

读完需要 6 分钟

速读仅需  2 分钟

/ 后台异步这一篇就够了 /

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):
    """
    发送提醒任务
    """
　　pass

2.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" % time.time()}

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。

我们启动项目看一下执行效果：

我坚信：思考问题的方法远大于具体解决问题的方案，让我们继续一路前行，下期再见！

原创不易，只愿能帮助那些需要这些内容的同行或刚入行的小伙伴，你的每次 点赞、分享 都是我继续创作下去的动力，我希望能在推广 python 技术的道路上尽我一份力量，欢迎在评论区向我提问，我都会一一解答，记得一键三连支持一下哦！

加入python学习交流微信群，请后台回复「入群」

往期推荐

python生产实战 python 闭包之庖丁解牛篇

大型fastapi项目实战 靠 python 中间件解决方案涨薪了

大型fastapi项目实战 高并发请求神器之aiohttp(下)

大型fastapi项目实战 高并发请求神器之aiohttp(上) [建议收藏]

本文分享自微信公众号 - python编程军火库（PythonCoder1024）。
如有侵权，请联系 support@oschina.cn 删除。
本文参与“OSC源创计划”，欢迎正在阅读的你也加入，一起分享。