FastAPI BackgroundTasks 后台任务完全指南在响应返回后执行的轻量任务机制【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi后台任务Background Tasks是 FastAPI 内置的一项轻量机制用于在向客户端发送响应之后再执行某些操作。本指南基于仓库中的官方教程 docs/ko/docs/tutorial/background-tasks.md结合 FastAPI 源码fastapi/background.py、fastapi/dependencies/utils.py、fastapi/routing.py与官方示例系统讲解后台任务的定义方式、依赖注入规则、底层运行原理及其与 Celery 等重型任务队列的取舍边界。读完本文你将掌握在路径操作函数与依赖中声明并合并后台任务、并判断何时该用或不该用BackgroundTasks。后台任务要解决什么问题在典型的 Web 请求处理中路径操作函数需要完成全部工作后才能返回响应客户端必须一直等待。但有一类操作发生在请求之后、且客户端无需等待其结果例如操作完成后的邮件通知连接邮件服务器并发送邮件通常需要数秒属于相对“慢”的操作。理想做法是立刻返回响应把发信放到后台执行避免用户白白等待。耗时数据处理例如接收一个需要长时间处理的文件可以先返回“Accepted”HTTP 202应答再在后台处理该文件。BackgroundTasks正是面向这类场景的官方内建方案它在响应已经发送给客户端之后才执行任务客户端的请求生命周期因此得以立即结束。值得强调的前提是FastAPI 的BackgroundTasks必须与响应发送绑定。从源码看路由处理中会把依赖求解阶段收集到的后台任务对象挂到响应对象上。在 fastapi/routing.py 中当路径操作函数返回一个Response实例且其background属性为空时框架会执行raw_response.background solved_result.background_tasks当函数返回的是普通数据JSON 等时则会在构造响应时把background一并传入fastapi/routing.py。也就是说只有任务被成功挂载到最终响应上它才会在响应发送后执行若请求处理中途抛出异常或响应未能发出则后台任务不会运行。快速上手三步注册一个后台任务第一步引入BackgroundTasks并声明参数完整示例见 docs_src/background_tasks/tutorial001_py310.pyfrom fastapi import BackgroundTasks, FastAPI app FastAPI() def write_notification(email: str, message): with open(log.txt, modew) as email_file: content fnotification 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, messagesome notification) return {message: Notification sent in the background}要点有两个从fastapi中导入BackgroundTasks注意类名末尾带有字母s在路径操作函数的参数中把某个参数的类型标注为BackgroundTasks。FastAPI 会替你创建这个BackgroundTasks对象并作为参数注入你不必手动实例化。这一步的底层逻辑位于依赖求解模块 fastapi/dependencies/utils.py只要参数树中存在后台任务参数名并且当前请求还没有创建过后台任务对象就执行background_tasks BackgroundTasks()随后将其注入到对应参数位置。第二步编写任务函数后台任务函数只是一个可以接收参数的普通 Python 函数。它既可以是async def异步函数也可以是普通def同步函数FastAPI 都知道如何正确处理——这一点与它对路径操作函数中async def/def的处理规则保持一致。由于示例中的文件写入操作并不需要使用async/await因此任务函数用普通def定义即可def write_notification(email: str, message): with open(log.txt, modew) as email_file: content fnotification for {email}: {message} email_file.write(content)这段代码模拟“发送一封邮件”把收件人和消息内容写入log.txt。真实场景中此处可替换为调用邮件 SDK、触发文件转码等任何耗时操作。第三步用.add_task()注册任务在路径操作函数内部通过后台任务对象的.add_task()方法把任务函数挂上去app.post(/send-notification/{email}) async def send_notification(email: str, background_tasks: BackgroundTasks): background_tasks.add_task(write_notification, email, messagesome notification) return {message: Notification sent in the background}.add_task()接收三类实参任务函数将在后台被调用的可调用对象本例为write_notification位置参数序列按顺序传给任务函数本例为email关键字参数以关键字形式传给任务函数本例为messagesome notification。当客户端POST到/send-notification/{email}时请求会立即返回 JSON{message: Notification sent in the background}而write_notification会等响应发送完毕后才执行将类似notification for email: some notification的内容写入文件。.add_task()的签名与语义直接继承自 Starlette 的同名实现FastAPI 在其上补充了类型标注与文档。在 fastapi/background.py 中可以看到add_task的func参数被明确标注为“响应发送后要调用的函数”且支持普通def与async def两种形态。在依赖注入中使用 BackgroundTasks同对象合并执行BackgroundTasks完整接入 FastAPI 的依赖注入系统。你可以在多个层级声明类型为BackgroundTasks的参数路径操作函数本身依赖函数dependable子依赖sub-dependency其他更深层的依赖。FastAPI 清楚在每个场景下该做什么、以及如何复用同一个对象从而保证在路径操作函数、依赖等各处添加的后台任务会被全部合并最终在响应发送后一起于后台依次执行。官方给出了跨依赖注入层的完整示例异步标注版本见 docs_src/background_tasks/tutorial002_an_py310.py使用Annotated与Depends同步写法版本见 docs_src/background_tasks/tutorial002_py310.pyfrom typing import Annotated from fastapi import BackgroundTasks, Depends, FastAPI app FastAPI() def write_log(message: str): with open(log.txt, modea) as log: log.write(message) def get_query(background_tasks: BackgroundTasks, q: str | None None): if q: message ffound 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 fmessage to {email}\n background_tasks.add_task(write_log, message) return {message: Message sent}运行逻辑拆解依赖函数get_query内部声明了background_tasks: BackgroundTasks参数若请求带有查询参数q则把found query: {q}\n作为一个后台日志任务添加路径操作函数send_notification自己也声明了background_tasks把message to {email}\n作为另一个后台日志任务添加由于两处注入的是同一个BackgroundTasks对象两个任务会被合并在响应{message: Message sent}发送之后按注册顺序先后追加写入log.txt。“跨层级共享同一个对象并合并任务”正是依赖求解器solve_dependencies的设计结果。函数签名中的background_tasks会贯穿整个依赖树的求解过程最终随SolvedDependency一并返回给路由层fastapi/dependencies/utils.py供 fastapi/routing.py 挂载到响应上。此外仓库中的测试 tests/test_dependency_contextmanager.py 亦验证了BackgroundTasks与依赖/上下文管理器协同工作时的合并行为可作为行为级佐证。技术细节BackgroundTasks 与 BackgroundTask 的区别BackgroundTasks类直接来源于 Starlette 的starlette.background模块。FastAPI 之所以把它重导出到自己的命名空间是为了让你可以直接from fastapi import BackgroundTasks避免误从starlette.background导入名称相近的替代类BackgroundTask末尾没有s。在 fastapi/background.py 中BackgroundTasks被定义为StarletteBackgroundTasks的子类并在 fastapi/init.py 中从fastapi.background模块对外导出。两者定位差异非常关键对比项BackgroundTasks复数BackgroundTask单数形态后台任务的集合容器单个后台任务对象声明方式作为路径操作函数参数FastAPI 自动注入需要手动BackgroundTask(func, *args)创建挂载方式FastAPI 自动处理需自行放进返回的 StarletteResponse中组合能力可容纳多个任务合并执行一个任务对应一个对象使用上的重要约定是当把BackgroundTasks用作路径操作函数参数时剩下的一切都由 FastAPI 接管这与直接注入Request对象的使用体验一致。FastAPI 在每次请求时自动实例化它、在依赖树中共享它、最后把它合并挂载到响应的background属性上。而单独使用BackgroundTask也是可行的只不过你必须在自己的代码中创建对象并返回一个显式包含该对象的 StarletteResponse例如from fastapi import FastAPI from fastapi.responses import JSONResponse from starlette.background import BackgroundTask def write_notification(email: str, message): with open(log.txt, modew) as email_file: content fnotification for {email}: {message} email_file.write(content) app FastAPI() app.post(/send-notification/{email}) async def send_notification(email: str): background_task BackgroundTask(write_notification, email, messagesome notification) return JSONResponse({message: Notification sent}, backgroundbackground_task)这种方式更底层、更灵活但需要你对响应对象与任务生命周期有更明确的控制。两种写法之间可按需求选择涉及多个任务、依赖注入协同优先使用 FastAPI 注入的BackgroundTasks只需在自定义响应上附加单个任务可考虑BackgroundTask。如需更深入的 Starlette 侧机制说明可继续阅读 Starlette 官方文档中关于 Background Tasks 的章节。重要注意事项什么时候不该用 BackgroundTasksBackgroundTasks适合“轻量 同进程”的场景但它并非万能的分布式任务方案。官方文档给出了非常清晰的边界如果需要进行重量级后台计算并且不一定要求任务与 FastAPI 应用运行在同一进程中例如不需要共享内存、变量等状态那么使用 Celery 这类更强大的任务工具会更合适Celery 这类方案通常需要更复杂的配置并依赖 RabbitMQ、Redis 等消息/任务队列管理器其优势是能够在多个进程、尤其是多台服务器上执行后台任务反之如果需要访问同一个 FastAPI 应用内的变量和对象或者只是执行小型后台任务例如发送一封邮件通知那么直接使用内置的BackgroundTasks就够了无需引入额外基础设施。一个实际的判断准则任务是否必须运行在当前应用进程内并访问应用状态需要 → 用BackgroundTasks不需要且规模较大、要求跨进程/跨服务器执行 → 引入 Celery 等队列方案。值得注意的是BackgroundTasks执行所依赖的函数与参数是在请求结束时随响应一起调度执行的它不提供任务重试、优先级、分布式调度等能力这些正是引入任务队列的典型动机。小结把后台任务机制沉淀为一句话即可复用的模式在路径操作函数或依赖中导入BackgroundTasks将其声明为参数FastAPI 自动注入再通过.add_task(func, *args, **kwargs)注册任务任务就会在响应发送之后于后台执行。回顾全篇的关键结论BackgroundTasks面向“响应后执行、客户端无需等待”的场景典型用途是邮件通知、数据后处理任务函数是普通函数支持async def与def两种写法底层执行时 FastAPI 会正确处理.add_task()接受任务函数、位置参数与关键字参数后台任务对象在依赖注入的多个层级间共享所有任务合并后在响应发送后执行源码依据见 fastapi/dependencies/utils.py 与 fastapi/routing.pyBackgroundTasks复数继承自 Starlette可在路径操作函数参数中直接注入单数的BackgroundTask需手动创建并随 StarletteResponse返回fastapi/background.py重计算、跨进程/跨服务器场景请选用 Celery 等任务队列轻量同进程任务用BackgroundTasks即可。与后台任务相关的可直接运行示例代码集中在 docs_src/background_tasks/其中tutorial001_py310.py为入门单任务版tutorial002_py310.py与tutorial002_an_py310.py为依赖注入合并多任务版可作为本地验证与二次开发的基础模板。【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考