三周精通FastAPI:29 定义在返回响应后运行的后台任务

官方文档:https://fastapi.tiangolo.com/zh/tutorial/background-tasks/

后台任务¶

你可以定义在返回响应后运行的后台任务。

这对需要在请求之后执行的操作很有用,但客户端不必在接收响应之前等待操作完成。

包括这些例子:

  • 执行操作后发送的电子邮件通知:
    • 由于连接到电子邮件服务器并发送电子邮件往往很“慢”(几秒钟),您可以立即返回响应并在后台发送电子邮件通知。
  • 处理数据:
    • 例如,假设您收到的文件必须经过一个缓慢的过程,您可以返回一个"Accepted"(HTTP 202)响应并在后台处理它。
from fastapi import BackgroundTasks, FastAPIapp = FastAPI()def write_notification(email: str, message=""):with open("log.txt", mode="w") as email_file:content = f"notification for {email}: {message}"email_file.write(content)@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):background_tasks.add_task(write_notification, email, message="some notification")return {"message": "Notification sent in the background"}

使用 BackgroundTasks

首先导入 BackgroundTasks 并在 路径操作函数 中使用类型声明 BackgroundTasks 定义一个参数:

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

FastAPI 会创建一个 BackgroundTasks 类型的对象并作为该参数传入。

创建一个任务函数¶

创建要作为后台任务运行的函数。

它只是一个可以接收参数的标准函数。

它可以是 async def 或普通的 def 函数,FastAPI 知道如何正确处理。

在这种情况下,任务函数将写入一个文件(模拟发送电子邮件)。由于写操作不使用 async 和 await,我们用普通的 def 定义函数:

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

添加后台任务¶

在你的 路径操作函数 里,用 .add_task() 方法将任务函数传到 后台任务 对象中:

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

.add_task() 接收以下参数:

  • 在后台运行的任务函数(write_notification)。
  • 应按顺序传递给任务函数的任意参数序列(email)。
  • 应传递给任务函数的任意关键字参数(message="some notification")。

依赖注入¶

使用 BackgroundTasks 也适用于依赖注入系统,你可以在多个级别声明 BackgroundTasks 类型的参数:在 路径操作函数 里,在依赖中(可依赖),在子依赖中,等等。FastAPI 知道在每种情况下该做什么以及如何复用同一对象,因此所有后台任务被合并在一起并且随后在后台运行:

from typing import Annotatedfrom fastapi import BackgroundTasks, Depends, FastAPIapp = 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: Annotated[str, Depends(get_query)]
):message = f"message to {email}\n"background_tasks.add_task(write_log, message)return {"message": "Message sent"}

该示例中,信息会在响应发出 之后 被写到 log.txt 文件。

如果请求中有查询,它将在后台任务中写入日志。

然后另一个在 路径操作函数 生成的后台任务会使用路径参数 email 写入一条信息。

技术细节¶

BackgroundTasks 类直接来自 starlette.background。

它被直接导入/包含到FastAPI以便你可以从 fastapi 导入,并避免意外从 starlette.background 导入备用的 BackgroundTask (后面没有 s)。

通过仅使用 BackgroundTasks (而不是 BackgroundTask),使得能将它作为 路径操作函数 的参数 ,并让FastAPI为您处理其余部分, 就像直接使用 Request 对象。

在FastAPI中仍然可以单独使用 BackgroundTask,但您必须在代码中创建对象,并返回包含它的Starlette Response

更多细节查看 Starlette's official docs for Background Tasks.

告诫¶

如果您需要执行繁重的后台计算,并且不一定需要由同一进程运行(例如,您不需要共享内存、变量等),那么使用其他更大的工具(如 Celery)可能更好。

它们往往需要更复杂的配置,即消息/作业队列管理器,如RabbitMQ或Redis,但它们允许您在多个进程中运行后台任务,甚至是在多个服务器中。

要查看示例,查阅 Project Generators,它们都包括已经配置的Celery。

但是,如果您需要从同一个FastAPI应用程序访问变量和对象,或者您需要执行小型后台任务(如发送电子邮件通知),您只需使用 BackgroundTasks 即可。

回顾¶

导入并使用 BackgroundTasks 通过 路径操作函数 中的参数和依赖项来添加后台任务。

实践

使用 BackgroundTasks创建任务

源代码

将代码写入background.py文件

from fastapi import BackgroundTasks, FastAPIapp = FastAPI()def write_notification(email: str, message=""):with open("log.txt", mode="w") as email_file:content = f"notification for {email}: {message}"email_file.write(content)@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):background_tasks.add_task(write_notification, email, message="some notification")return {"message": "Notification sent in the background"}

启动服务

uvicorn background:app --reload

 测试

使用命令:

curl -X 'POST' \'http://127.0.0.1:8000/send-notification/test%40example.com' \-H 'accept: application/json' \-d ''

输出:

{"message":"Notification sent in the background"}

查看log.txt文件:

cat log.txt 
notification for test@example.com: some notification(base)

依赖注入

源代码

代码写入backgroundtask.py文件:

from typing import Annotatedfrom fastapi import BackgroundTasks, Depends, FastAPIapp = 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: Annotated[str, Depends(get_query)]
):message = f"message to {email}\n"background_tasks.add_task(write_log, message)return {"message": "Message sent"}

启动服务

uvicorn backgroundtask:app --reload

测试

curl -X 'POST' \'http://127.0.0.1:8000/send-notification/test%40example.com?q=x%3D1%26b%3D888' \-H 'accept: application/json' \-d ''

输出:

{"message":"Message sent"}

查看log.txt文件:

cat log.txt 
found query: x=1&b=888
message to test@example.com

总结

使用background_tasks.add_task,将要执行的函数放进去即可

    background_tasks.add_task(write_notification, email, message="some notification")

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.rhkb.cn/news/463035.html

如若内容造成侵权/违法违规/事实不符,请联系长河编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

Rust的enum枚举的强大用法

在Rust中,enum(枚举)是一种非常强大的类型,它可以包含多个变体(variants),每个变体可以是不同的类型,包括复杂类型。这使得enum在Rust中不仅用于表示简单的状态或选项集合&#xff0…

Python世界:自动化办公Word之批量替换文本生成副本

Python世界:自动化办公Word之批量替换文本生成副本 任务背景编码思路代码实现相关参考 任务背景 为提高办公效率,用python试手了一个word任务,要求如下: 给你一个基础word文档A,格式为docx,名字为&#xf…

Git下载-连接码云-保姆级教学(连接Gitee失败的解决)

Git介绍 码云连接 一、Git介绍 二、Git的工作机制 下载链接:Git - 下载软件包 三、使用步骤 创建一个wss的文件夹,作为‘工作空间’ 四、连接码云账号 五、连接Gitee失败的解决方法 一、Git介绍 Git是一个免费的、开源的分布式版本控制…

【JavaEE初阶】网络原理(4)

欢迎关注个人主页:逸狼 创造不易,可以点点赞吗~ 如有错误,欢迎指出~ 目录 网络层 > IP协议 IP协议报头结构 4位版本 4位首部长度 8位服务类型(TOS) 16位总长度(字节数), 16位标识 3位标志位 13位片偏移 8位生存时间(TTL) 8位协议 16位首部…

vue3+vant实现视频播放(含首次禁止进度条拖拽,视频看完后恢复,保存播放视频进度,刷新及下次进入继续播放,判断视频有无全部看完等)

1、效果图 2、 <div><videocontrolsclass"video_player"ref"videoPlayer":src"videoSrc"timeupdate"handleTimeUpdate"play"onPlay"pause"onPause"ended"onVideoEnded"></video><…

LongVU :Meta AI 的解锁长视频理解模型,利用自适应时空压缩技术彻底改变视频理解方式

Meta AI在视频理解方面取得了令人瞩目的里程碑式成就&#xff0c;推出了LongVU&#xff0c;这是一种开创性的模型&#xff0c;能够理解以前对人工智能系统来说具有挑战性的长视频。 研究论文 "LongVU&#xff1a;用于长视频语言理解的时空自适应压缩 "提出了一种革命…

【Spring IOC】实现一个简单的 Spring 容器

1. 理解知识 Spring 容器的作用 Spring 容器是Spring的核心&#xff0c;用来管理Bean对象的。容器将创建对象&#xff0c;把它们连接在一起&#xff0c;配置它们&#xff0c;并管理他们的整个生命周期从创建到销毁。 Bean 对象的管理 当一个 Bean 对象交给 Spring 容器管理…

Pytorch lightning多机多卡训练通讯问题(NCCL error)排查

一、问题 单机多卡可以正常训练模型&#xff0c;多机多卡数据加载完成后卡住不动&#xff0c;排查两台机器可以ping通&#xff0c;表明网络没有问题&#xff0c;查看bug信息是NCCL通信问题。报错信息大致如下: torch.distributed.DistBackendError: NCCL error in: …/torch/c…

攻防靶场(26):hydra爆破web的小技巧 DC-4

目录 1. 侦查 1.1 收集目标网络信息&#xff1a;IP地址 1.2 主动扫描&#xff1a;扫描IP地址段 1.3 主动扫描&#xff1a;字典扫描 1.4 主动扫描&#xff1a;漏洞扫描 2. 初始访问 2.1 有效账户&#xff1a;默认账户 2.2 利用面向公众的应用 3. 凭据访问 3.1 不安全的凭据&…

UOS 安装usb wifi 网卡驱动

电脑上装安uos后发现usb网卡驱动不见了&#xff0c;网卡长下面这个样子&#xff0c;但是官方没有驱动 驱动网址选5300 https://www.ezcast.com/app/ezcast/wifi-adapter/windows 这时我们 lsusb找到相关设备&#xff0c;发现是Realtek 的设备 要在 Ubuntu 上安装 Realtek 0bda…

东方娱乐周刊

文章目录 一、征稿简介二、重要信息三、服务简述四、投稿须知五、联系咨询 一、征稿简介 二、重要信息 期刊官网&#xff1a;https://ais.cn/u/3eEJNv 三、服务简述 学科领域&#xff1a; 人文社科-教育学、文学、艺术、体育、人文社科&#xff1a;其他 四、投稿须知 1.在…

第三方支付系统架构设计

第三方支付是指具备一定实力和信誉保障&#xff0c;并获得国家颁发运营拍照的独立机构&#xff0c;采用和各大银行签约的方式&#xff0c;通过与银行相关接口对接而促成交易的网络支付的模式。我们熟悉的微信支付和支付宝都属于第三方支付工具&#xff0c;第三方支付工具随着移…

计算机网络八股文个人总结

1.TCP/IP模型和OSI模型的区别 在计算机网络中&#xff0c;TCP/IP 模型和 OSI 模型是两个重要的网络协议模型。它们帮助我们理解计算机通信的工作原理。以下是它们的主要区别&#xff0c;以通俗易懂的方式进行解释&#xff1a; 1. 模型层数 OSI 模型&#xff1a;有 7 层&#…

Java日志脱敏(二)——fastjson Filter + 注解 + 工具类实现

背景简介 日志脱敏 是常见的安全需求&#xff0c;最近公司也需要将这一块内容进行推进。看了一圈网上的案例&#xff0c;很少有既轻量又好用的轮子可以让我直接使用。我一直是反对过度设计的&#xff0c;而同样我认为轮子就应该是可以让人拿去直接用的。所以我准备分享两篇博客…

高效实现SCRM用户管理的最佳实践与策略

内容概要 在当今竞争激烈的市场环境中&#xff0c;SCRM用户管理显得尤为重要。SCRM&#xff08;Social Customer Relationship Management&#xff09;不仅仅是简单的客户管理工具&#xff0c;它更是企业与客户之间建立良好关系的一座桥梁。通过深入了解用户的需求和行为&…

Git - 两种方式撤销已提交到远端仓库的记录并删除提交记录

文章目录 命令行方式附 命令行方式 确定要撤销的提交记录 首先&#xff0c;使用以下命令查看提交历史&#xff1a; git log找到想撤销的提交记录的哈希值&#xff08;SHA&#xff09; &#xff0c;比如9c9c98d6f7f28c41d971f8efd51ed31f9720792c 撤销提交记录 根据需求选择以下…

【C/C++】字符/字符串函数(0)(补充)——由ctype.h提供

零.导言 除了字符分类函数&#xff0c;字符转换函数也是一类字符/字符串函数。 C语言提供了两种字符转换函数&#xff0c;分别是 toupper &#xff0c; tolower。 一.什么是字符转换函数&#xff1f; 顾名思义&#xff0c;即转换字符的函数&#xff0c;如大写字母转小写字母&am…

【排序】5.堆排序(详细图解)

文章目录 前言1.建堆方法的选择2.优先使用向下调整的原因3.堆排序图解&#xff08;大堆-升序为例子&#xff09;3.1 向下调整法-建大堆3.2 进行堆排序 4.堆排序代码4.1 向下调整法4.1. 1 小堆4.1. 2 大堆4.2 堆排序 5. 关于小堆——降序6.性能分析 前言 &#x1f431;个人主页&…

头歌——机器学习(线性回归)

文章目录 线性回归简述答案 线性回归算法答案 线性回归实践 - 波斯顿房价预测LinearRegression代码 利用sklearn构建线性回归模型示例代码如下&#xff1a; 代码 线性回归简述 简单线性回归 在生活中&#xff0c;我们常常能碰到这么一种情况&#xff0c;一个变量会跟着另一个变…

技术美术百人计划 | 《5.4 水体渲染》笔记

一、水体渲染的波形模拟技术-基于物理 基于物理的波形模拟方法&#xff1a; 欧拉方法&#xff08;Eulerian approaches&#xff09;[Kass 1990]拉格朗日方法&#xff08;Lagrangian approaches&#xff09; [Stam 1995]欧拉-拉格朗日混合方法&#xff08;Hybrid approaches&a…