fastapi_No.18_后台应用

简介

后台应用就是在发送响应后运行的函数。通常用在请求后需要耗费较长时间处理，且客户端不需要在接收请求前接收处理结果。也就是说后台应用可以理解为两点：

• 处理时间长
• 客户端对该函数响应结果无要求

使用场景

• 邮件提醒
• 数据处理

创建后台应用步骤

第一步：定义后台应用函数

# 创建一个函数，表示后台运行的函数
# 定义后台运行函数
def send_email(email:str,message:str):
    with open("log.txt",mode='a+') as email_file:
        content = f"notification for {email}: {message}"
        email_file.writelines(content)
1
2
3
4
5
6

第二步：使用后台应用函数

从fastapi中导入BackgroundTasks，并在路径操作函数中定义一个BackgroundTasks类型的参数。

#导入BackgroundTasks类
from fastapi import BackgroundTasks, FastAPI
app = FastAPI()
@app.post("/send-notification/{email}")
#在路径操作函数中添加BackgroundTasks类的参数
async def send_notification(email:str, bt:BackgroundTasks):
    # 将后台运行函数，添加到BackgroundTasks实例对象中，并传入后台运行函数的形参值
    bt.add_task(send_email,email,message=datetime.now().isoformat())
    return {"message":"Notification sent in the background"}
1
2
3
4
5
6
7
8
9

完整代码

# 导入包
from fastapi import FastAPI,BackgroundTasks
from datetime import datetime
app = FastAPI()
# 定义后台运行函数
def send_email(email:str,message:str):
    with open("log.txt",mode='a+') as email_file:
        content = f"notification for {email}: {message}"
        email_file.writelines(content)
@app.post("/send-notification/{email}")
#在路径操作函数中添加BackgroundTasks类的参数
async def send_notification(email:str, bt:BackgroundTasks):
    # 将后台运行函数，添加到BackgroundTasks实例对象中，并传入后台运行函数的形参值
    bt.add_task(send_email,email,message=datetime.now().isoformat())
    return {"message":"Notification sent in the background"}
if __name__ == "__main__":
    import uvicorn
    uvicorn.run("singlebackgroundtask:app",host="127.0.0.1",port=8080,reload=True)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

测试结果：

在依赖项中的后台应用

后台应用除了可以应用在路径操作函数中外，还可以应用在依赖项中，也可以应用在子依赖中。
fastapi不是每一步在每个有后台应用的函数运行结束后就立刻运行后台应用函数，而是在说有的函数都运行结束后，fastapi根据之前记录的调用后台应用函数的顺序，一次执行多次后台应用函数。

示例代码

这里编写了一个示例来说明这个问题：

from typing import Union
import time
from fastapi import BackgroundTasks,Depends,FastAPI
app = FastAPI()
# 定义一个后台应用函数
def write_log(message:str):
    with open("log1.txt",mode="a") as log:
        log.write(message+'\n')

# 定义一个嵌套的后台应用函数
def get_query(background_tasks:BackgroundTasks,q:Union[str,None] = None):
    if q:
        message = f"found query: {q}"
        background_tasks.add_task(write_log,message)
    print("依赖函数执行完毕")
    return q
@app.post("/send-notification/{email}")
async def send_notification(
    email:str, background_tasks:BackgroundTasks,q:str = Depends(get_query)
):
    message = f"message to {email}"
    background_tasks.add_task(write_log,message)
    print("路径操作函数执行完毕")
    time.sleep(5)
    return {"message":"message sent","q":q}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("multibackgroundtask:app",host="127.0.0.1",port=8080,reload=True)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

由于让系统睡眠了5s，通过观察目前已经打印了函数执行完毕。但是还没有后台运行程序执行，因为文件中还没有log1.txt文件出现。
这就说明了后台应用函数是一起执行的，且晚于依赖函数和路径操作函数。

后台应用函数输出文件内容看出，先执行了依赖函数中的那次输出，在执行了路径操作函数中的那次输出。

示例代码说明

参考下图进行理解。