07-11 09:45 阅读 91

FastAPI后台任务

一、后台任务使用　　

你可以定义后台任务在后台响应之后继续运行，这对于在请求之后去做一些操作时有用的，但是客户端不会真正的等待响应中操作的完成。这包括，例如：

执行操作后发送电子邮件通知
处理数据，比如，后台接收一个文件需要处理，但是可以先给客户端返回响应，然后后台接着处理

1、使用后台任务

首先，导入BackgroundTask以及在路径操作函数中定义一个参数，并且声明类型为BackgroundTasks：

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    pass

FastAPI将会创建BackgroundTasks类型的对象，并且将其当作参数传递。

2、创建一个任务函数

创建一个函数作为后台任务，它是一个可以接收参数的标准函数，可以使用async异步方式或者正常函数方式，FastAPI知道应该如何使用它，在这个例子中，这个后台任务将会写一个文件：

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def write_notification(email: str, message=""):
    with open("log.txt", mode="w") as f:
        f.write(f"notification for {email}:{message}")


@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    pass

3、添加后台任务

在路径操作函数中，通过".add_task()"函数将任务函数添加到后台任务对象中：

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def write_notification(email: str, message=""):
    with open("log.txt", mode="w") as f:
        f.write(f"notification for {email}:{message}")


@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, message="notification...")
    return {"message": "Notification sent in the background"}

.add_task()接收的参数：

一个在后台运行的任务函数（write_notification）
按照顺寻传递的一系列参数（email）
任何的关键字参数（message="notification..."）

二、依赖注入

在依赖注入系统中也可以使用后台任务，你可以声明一个BackgroundTasks类型的参数，在路径操作函数、依赖项、子依赖项中等。

from fastapi import BackgroundTasks, FastAPI, Depends
from typing import Optional

app = FastAPI()


def write_log(message: str):
    with open("log.txt", mode="a") as f:
        f.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("/dependency/background_tasks")
async def write_log(q: str = Depends(get_query)):
    if q:
        return {"message": "write log in the background"}

原文：https://www.cnblogs.com/shenjianping/p/14873320.html