标签：__ Task FastAPI email task background 后台任务 message
FastAPI（41）- Background Task 后台任务
后台任务
- 顾名思义，可以在返回响应后运行后台任务
- 这对于需要在请求后执行特定的操作很有用，且客户端并不需要在接收响应之前等待该操作完成
常见的栗子
- 发送电子邮件通知，由于连接到电子邮件服务器并发送电子邮件往往会比较“缓慢”（几秒钟），因此可以立即返回响应并在后台发送电子邮件通知
- 假设您到一个必须经过缓慢处理的文件，可以先返回“已接受”（HTTP 202）响应并在后台处理它
实际栗子
创建后台任务要用到的函数
- 创建一个作为后台任务运行的函数，就是一个普通函数
- 可以加 async 也可以不加，FastAPI 将会正确处理它
import time
def write_notification(email: str, message: str = ""):
# 1、模拟和邮件服务器建立连接
time.sleep(3)
with open("text.txt", mode="w") as f:
# 2、模拟发送邮件
content = f"message is {message}"
f.write(content)
print(content)
添加后台任务
#!usr/bin/env python
# -*- coding:utf-8 _*-
"""
# author: 小菠萝测试笔记
# blog: https://www.cnblogs.com/poloyy/
# time: 2021/9/29 7:11 下午
# file: 35_background_task.py
"""
import time
import uvicorn
from fastapi import FastAPI, BackgroundTasks
app = FastAPI()
def write_notification(email: str, message: str = ""):
# 1、模拟和邮件服务器建立连接
time.sleep(3)
with open("text.txt", mode="w") as f:
# 2、模拟发送邮件
content = f"message is {message}"
f.write(content)
print(content)
@app.post("/email/{email}")
async def send_email(
email: str,
# 指定参数类型为 BackgroundTasks
background_task: BackgroundTasks
):
# 添加后台任务
# 第一个参数：可调用对象，一般就是函数
# 后面的参数：write_notification 函数所需要的参数
background_task.add_task(write_notification, email, message="test_message")
return {"message": "Notification sent in the background"}
if __name__ == '__main__':
uvicorn.run(app="35_background_task:app", reload=True, host="127.0.0.1", port=8080)
后台任务结合依赖项
#!usr/bin/env python
# -*- coding:utf-8 _*-
"""
# author: 小菠萝测试笔记
# blog: https://www.cnblogs.com/poloyy/
# time: 2021/9/29 7:11 下午
# file: 35_background_task.py
"""
import time
from typing import Optional
import uvicorn
from fastapi import FastAPI, BackgroundTasks, Depends
app = FastAPI()
# 后台任务函数
def write_log(message: str):
with open("log.txt", mode="a") as log:
log.write(message)
# 依赖项函数
async def get_query(
background_task: BackgroundTasks,
q: Optional[str] = None,
):
# 如果 q 有值才执行后台任务
if q:
message = f"found query: {q}\n"
background_task.add_task(write_log, message)
@app.post("/email_depends/{email}")
async def send_email(
email: str,
background_task: BackgroundTasks,
q: str = Depends(get_query)
):
# 执行一次后台任务
message = f"message to {email}\n"
background_task.add_task(write_log, message)
return {"message": "Message sent"}
if __name__ == '__main__':
uvicorn.run(app="35_background_task:app", reload=True, host="127.0.0.1", port=8080)
- 后台任务可以在任意地方使用，比如路径操作、依赖项、子依赖项...
- FastAPI 会将所有后台任务合并在一起，然后在后台会按 add_task 的顺序运行
查看 BackgroundTasks 源码
- BackgroundTasks 是继承 BackgroundTask，而 BackgroundTask 是直接来自 starlette.background
- add_task() 第一个参数 func 类型是 Callable，可调用对象，一般传函数就好啦
- 内部会声明一个 BackgroundTask 对象，自动调用它的 __call__ 方法
- 可以看到，最终会执行 func()
- func() 函数参数就是 add_task() 函数除第一个参数以外的参数
BackgroundTasks 注意事项
- 如果需要执行繁重的后台计算，且可能需要多个进程运行（例如，不需要共享内存、变量等），使用其他更大的工具，如：Celery，效果可能会更好
- 它们往往需要更复杂的配置、消息/作业队列管理器，如 RabbitMQ 或 Redis，它们允许在多个进程中运行后台任务，尤其是在多个服务器中
- 但是，如果需要从同一个 FastAPI 应用程序访问变量和对象，或者需要执行小型后台任务（例如发送电子邮件通知），只需使用 BackgroundTasks
标签：__,Task,FastAPI,email,task,background,后台任务,message 来源： https://www.cnblogs.com/xiao-xue-di/p/15770160.html
本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享； 2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关； 3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关； 4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除； 5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。