后台任务

• 后台任务
  • 与依赖注入一起使用

---

【独家新技术】从0到1学习 FastAPI 框架的所有知识点_哔哩哔哩_bilibili

Background Tasks - FastAPI (tiangolo.com)

最典型的使用是: 用户注册之后发邮件

用户能够在前端立刻得到返回, 但是接口中实行的是比较耗时的任务

引入 fastapi.BackgroundTask 后通过在异步函数中调用其中的 add_task 来添加后台任务

##### Background Tasks 后台任务 #####
import os
from fastapi import APIRouter, BackgroundTasks, Depends

file_path = os.path.abspath(os.path.join(os.path.dirname(__file__), './README.md'))

def bg_task(framework: str):
    """已续写的形式用 utf-8 编码写入README.md"""
    with open(file_path, mode="a", encoding="utf-8") as f:
        f.write(f"\n## {framework} 框架精讲")


@app08.post("/background_tasks")
async def run_bg_task(framework: str, background_tasks: BackgroundTasks):
    """
    :param framework: 被调用的后台任务函数的参数
    :param background_tasks: FastAPI.BackgroundTasks
    :return:
    """
    background_tasks.add_task(bg_task, framework)
    return {"message": "任务已在后台运行"}


def continue_write_readme(background_tasks: BackgroundTasks, q: Optional[str] = None):
    if q:
        background_tasks.add_task(bg_task, 
        "\n> 整体的介绍 FastAPI, 快速上手开发, 结合 API 交互文档逐个讲解核心模块的使用\n")
    return q


@app08.post("/dependency/background_tasks")
async def dependency_run_bg_task(q: str = Depends(continue_write_readme)):
    """用依赖注入的方式导入后台任务
    """
    if q:
        return {"message": "README.md更新成功"}

---

与依赖注入一起使用

Background Tasks - FastAPI (tiangolo.com)

---

使用 BackgroundTasks 还可以与依赖注入系统一起工作, 你可以在多个层次上声明一个 BackgroundTasks 类型的参数(可以在 path operation 函数中, 在 dependency(dependable) 中, 亦可以在 sub-dependency 等处声明)

Python 3.10 and above

# Python 3.10 and above
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: str | None = None):
    if q:
        message = f"found query: {q}\n"
        background_tasks.add_task(write_log, message)
    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}\n"
    background_tasks.add_task(write_log, message)
    return {"message": "Message sent"}

在这个示例中 query 参数传入 email 和 q

接口在处理完 email 生成 message 并返回给用户后会将 message 传给后台任务 weite_log 来记录日志

如果 query 参数中有 q, 那么它会在 get_query 函数中处理然后创给后台任务 write_log 来记录日志

---