首页
/ FastAPI Lifespan 事件(startup/shutdown)深入解析:用 lifespan 参数优雅管理应用启动与关闭逻辑

FastAPI Lifespan 事件(startup/shutdown)深入解析:用 lifespan 参数优雅管理应用启动与关闭逻辑

2026-09-07 15:06:17作者:凤尚柏Louis

导读:本指南围绕仓库 docs/hi/docs/advanced/events.md(印地语版,其英文原版见 docs/en/docs/advanced/events.md)展开,讲解如何在 FastAPI 应用开始接收请求之前执行一次性初始化代码(如加载机器学习模型、创建数据库连接池),并在应用处理完请求、停止之前执行清理代码(如释放内存/GPU、关闭连接池)。读完本文,你将掌握现代推荐方案 lifespan(async context manager)的完整写法、底层执行原理,以及已被废弃的 startup/shutdown 事件处理器为何不再被推荐。

为什么需要"应用级生命周期"代码

FastAPI 允许你定义两类特殊逻辑:

  • startup 逻辑:在应用启动时执行。它会在应用开始接收请求之前,恰好执行 一次
  • shutdown 逻辑:在应用关闭时执行。它会在应用可能处理了大量请求之后,恰好执行 一次

因为这段代码执行于应用"开始接单"之前与"处理完毕"之后,所以它完整覆盖了整个应用的生命周期(lifespan)。这一点非常适合初始化整个应用共享、跨请求复用、且事后需要清理的资源,例如:

  • 数据库连接池(database connection pool);
  • 需要整体加载、可在所有请求间复用的共享机器学习模型;
  • 缓存、消息队列客户端、外部 SDK 客户端等重量级单例。

典型场景:共享 ML 模型的"昂贵加载"

设想你有若干机器学习模型,需要用它处理请求,并且这些模型在所有请求之间共享(并非每请求一个、每用户一个)。模型加载通常很慢——需要从磁盘读取大量数据,因此绝不能为每个请求都加载一遍。

最直观的做法是在模块/文件的顶层加载模型:

# 顶层加载:问题在于即使只跑一个简单的自动化测试,也会触发模型加载
model = load_big_model_from_disk()

这样带来的副作用是:哪怕你只是想跑一条独立的单元测试,测试进程也会因为被迫等待模型加载而变得缓慢

lifespan 正是为了解决这个问题而设计的:让模型在代码被 import 时不被加载,而是在应用真正开始接收请求的瞬间才加载。

现代推荐方案:FastAPI(lifespan=...) + async context manager

你可以在 FastAPI 应用实例化时传入 lifespan 参数,并结合 Python 的 context manager 机制定义 startupshutdown 逻辑。

完整可运行示例

仓库中的官方示例位于 docs_src/events/tutorial003_py310.py

from contextlib import asynccontextmanager

from fastapi import FastAPI


def fake_answer_to_everything_ml_model(x: float):
    return x * 42


ml_models = {}


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Load the ML model
    ml_models["answer_to_everything"] = fake_answer_to_everything_ml_model
    yield
    # Clean up the ML models and release the resources
    ml_models.clear()


app = FastAPI(lifespan=lifespan)


@app.get("/predict")
async def predict(x: float):
    result = ml_models"answer_to_everything"
    return {"result": result}

代码语义一目了然:

  • yield 之前的代码:模拟昂贵的启动操作——把(假的)模型函数放进共享字典 ml_models。这段代码在应用开始接收请求之前startup 阶段)执行一次;
  • yield 之后的代码:执行清理,例如清空模型、释放内存或 GPU 资源。这段代码在应用处理完所有请求之后、正式关闭之前执行一次;
  • 文档还特别提示:shutdown 通常发生在我们主动停止应用之时(比如要发布新版本,或只是不想再运行它了)。

lifespan 函数:一个带 yield 的 async 函数

首先要注意,lifespan() 是一个包含 yieldasync def 函数。这与 FastAPI 的"带 yield 的依赖(Dependencies with yield)"在结构上非常相似:yield 把函数一分为二,前半段在应用启动前执行,后半段在应用结束后执行。

Async Context Manager:@asynccontextmanager 做了什么

示例中的 lifespan()@asynccontextmanager 装饰(来自 Python 标准库 contextlib),它把普通函数转换为所谓的 async context manager

  • 在 Python 中,context manager 就是可以在 with 语句中使用的对象,例如 open()
with open("file.txt") as file:
    file.read()
  • 在新版本 Python 中还有 async context manager,配合 async with 使用:
async with lifespan(app):
    await do_stuff()

context manager 的执行约定是:进入 with 代码块前,先执行 yield 之前的代码;退出 with 代码块后,执行 yield 之后的代码。在我们的示例中并没有直接手动调用它,而是把它交给 FastAPI,由 FastAPI 在合适的时机(ASGI server 启动/关闭时)自动进入与退出。

一句话总结FastAPIlifespan 参数接收的正是这样一个 async context manager,所以我们把上面写好的 lifespan 直接传给 FastAPI(lifespan=lifespan) 即可。

从源码看 FastAPI 如何消费 lifespan 参数

lifespan 并不仅是文档层面的约定,仓库源码对其有清晰的实现支撑。

  1. 应用入口透传:在 fastapi/applications.py 中,FastAPI.__init__ 的类型注解与文档说明均明确指出 on_startup/on_shutdown 应改用 lifespan,而 lifespan 参数被描述为 "A Lifespan context manager handler. This replaces startup and shutdown functions with a single context manager.";随后该参数在构造内部路由时被透传给 APIRouter(见 fastapi/applications.py)。

  2. 参数归一化处理:在 fastapi/routing.py 中,APIRouter 会对传入的 lifespan 做归一化:

if lifespan is None:
    # Use the default lifespan that runs on_startup/on_shutdown handlers
    lifespan_context: Lifespan[Any] = _DefaultLifespan(self)
elif inspect.isasyncgenfunction(lifespan):
    lifespan_context = asynccontextmanager(lifespan)
elif inspect.isgeneratorfunction(lifespan):
    lifespan_context = _wrap_gen_lifespan_context(lifespan)
else:
    lifespan_context = lifespan
self.lifespan_context = lifespan_context

从这段源码可以推断出三点内部约定:

  • 如果你用 @asynccontextmanager 装饰 lifespan(即传入 async generator function),它会原样被 asynccontextmanager 包装,yield 前后两段代码分别成为 ASGI lifespan 协议的启动与关闭处理;
  • 若传入的是普通(同步)生成器函数,会走 _wrap_gen_lifespan_context 兼容分支;
  • 若什么都不传,则使用内置的 _DefaultLifespan 去调度旧式的 on_startup/on_shutdown 处理器。
  1. 向后兼容的默认实现_DefaultLifespan 定义在 fastapi/routing.py。其源码注释说明,这是对 Starlette 已移除的 _DefaultLifespan 的复刻,FastAPI 保留它是为了继续兼容 on_startup/on_shutdown 事件处理器。它通过 __aenter__ 调用 self._router._startup()、通过 __aexit__ 调用 self._router._shutdown(),从而把旧式事件处理器桥接到新的 lifespan 机制上。

也就是说,无论你使用哪种写法,最终都会收敛到 ASGI 的 lifespan 执行管线中,只是入口不同。

已被废弃的替代方案:startup / shutdown 事件处理器

lifespan 成为主流之前,FastAPI 提供过一套事件处理器写法,对应代码入口为 @app.on_event(...)官方已在文档中明确标记为 deprecated,并强烈建议跳过此节、改用 lifespan

⚠️ 关键警告(原文照录):处理 startup / shutdown 的推荐方式是使用上述 FastAPIlifespan 参数。一旦提供了 lifespan 参数,startupshutdown 事件处理器将不再被调用。二者只能选其一:要么全部交给 lifespan,要么全部用旧式事件,绝不可混用。

这一说法同样可以在源码中得到印证:在 fastapi/applications.py 中,on_event 方法的文档字符串明确写着 "on_event is deprecated, use lifespan event handlers instead.";对应的测试文件也通过 pytest.warns(DeprecationWarning) 来断言注册事件会触发弃用警告(见 tests/test_tutorial/test_events/test_tutorial001.py)。

旧式 startup 事件

要注册一个在应用启动前运行的函数,可以用事件名 "startup" 声明(官方示例见 docs_src/events/tutorial001_py310.py):

from fastapi import FastAPI

app = FastAPI()

items = {}


@app.on_event("startup")
async def startup_event():
    items["foo"] = {"name": "Fighters"}
    items["bar"] = {"name": "Tenders"}


@app.get("/items/{item_id}")
async def read_items(item_id: str):
    return items[item_id]

在上例中,startup 处理器把 items "数据库"(其实只是一个 dict)预置为若干值。旧式事件模型支持注册多个事件处理函数,并且应用只有在所有 startup 处理器都执行完毕之后才会开始接收请求。

旧式 shutdown 事件

要注册一个在应用关闭时运行的函数,则用事件名 "shutdown"(官方示例见 docs_src/events/tutorial002_py310.py):

from fastapi import FastAPI

app = FastAPI()


@app.on_event("shutdown")
def shutdown_event():
    with open("log.txt", mode="a") as log:
        log.write("Application shutdown")


@app.get("/items/")
async def read_items():
    return [{"name": "Foo"}]

这里 shutdown 处理器会向 log.txt 写入一行 "Application shutdown"。原文档特别做了两点说明:

  • open()mode="a" 表示 append(追加):新行会加在文件已有内容之后,而不会覆盖旧内容;
  • 这里交互的是磁盘文件,涉及 I/O 等待,但标准 open() 并不使用 async/await,所以该事件处理函数用普通 def 声明即可,无需写成 async def。这体现了一条通用选型原则:只有确实需要 await 的阻塞型 I/O 逻辑才使用 async def,纯同步操作(或同步库封装)用普通 def 更合适。

为什么旧式"分开写"不理想:startup 与 shutdown 通常成对出现

实际项目中,startupshutdown 的逻辑高度关联:通常是"启动某资源 → 结束后关闭它"、"acquire 一个资源 → 再 release 它"。若用两个互不共享逻辑/变量的独立函数实现,就必须把状态暂存到全局变量之类的取巧手段里,代码可读性与封装性都会变差。

正是出于这个原因,官方文档现在推荐改用前文介绍的 lifespan:把成对的初始化与清理写进同一个函数,紧邻 yield 前后,资源共享自然、逻辑一目了然。

底层技术细节:ASGI Lifespan 协议

从技术实现上看,这套机制并不属于 HTTP 层,而是 ASGI 规范的一部分。在 ASGI 技术规范中,它对应 Lifespan Protocol,定义了名为 startupshutdown 的事件:

  • ASGI server(如 Uvicorn)启动应用时,会向应用发送 lifespan startup 消息,FastAPI 借此执行 yield 之前(或旧式 startup 处理器)的代码;
  • server 即将退出时,会发送 lifespan shutdown 消息,FastAPI 借此执行 yield 之后(或旧式 shutdown 处理器)的代码。

因此,FastAPI 应用的这些生命周期代码并不由应用自身调用,而是由承载它的 ASGI 服务器在合适的时机驱动。如果你希望深入了解 lifespan state 如何在代码其他区域共享,可以继续阅读仓库中的相关文档;本仓库英文原版文档在 docs/en/docs/advanced/events.md 的技术细节一节中还提到了可参考 Starlette 的 lifespan 文档(包括 lifespan state 的处理方式),此处不再展开。

重要限制:事件只对主应用生效,不作用于 Mounts 子应用

原文档最后用 🚨 强调了一个极易踩坑的细节:

这些 lifespan 事件(startup / shutdown)只会对 main(主)应用执行,而不会为通过 mount 挂载的 **Sub Applications(子应用)**执行。

也就是说,如果业务通过 app.mount() 把另一个独立的 FastAPI/ASGI 应用挂载为主应用的子路径(实现各自独立的 OpenAPI 与文档 UI),那么子应用自己注册的 lifespan / 事件逻辑并不会随主应用启动而运行。关于如何创建与挂载子应用,可阅读 docs/en/docs/advanced/sub-applications.md(即原文档中 sub-applications.md 链接在仓库中的位置,对应文档讲解 "Sub Applications - Mounts")。在组织多应用架构时,应把需要执行生命周期初始化的逻辑放在顶层主应用中,或为子应用安排独立的运行方式。

用测试验证 lifespan 的执行时机

仓库中配套的官方测试能直观证明 lifespan 的执行顺序,也适合作为自定义测试的范式。

tests/test_tutorial/test_events/test_tutorial003.py 中:

def test_events():
    assert not ml_models, "ml_models should be empty"
    with TestClient(app) as client:
        assert ml_models["answer_to_everything"] == fake_answer_to_everything_ml_model
        response = client.get("/predict", params={"x": 2})
        assert response.status_code == 200, response.text
        assert response.json() == {"result": 84.0}
    assert not ml_models, "ml_models should be empty"

测试验证了三个关键点:

  1. 进入 with TestClient(app) as client: 之前ml_models 为空——lifespan 尚未执行;
  2. 进入 with 块(即 TestClient 触发完整 lifespan 启动)后,ml_models["answer_to_everything"] 已就绪,且 GET /predict?x=2 返回 {"result": 84.0}——说明 yield 之前的初始化已在应用接收请求前完成,并在路径操作中真实可用;
  3. 退出 with 块后 ml_models 再次为空——说明 yield 之后的清理代码在应用关闭阶段被准确执行。

这正是 lifespan 较"模块顶层加载"更优的直接证据:在跑不依赖该模型的其他测试时,根本不会触发模型加载,测试不会变慢。

结语:何时用哪种方案

需求 推荐做法 说明
初始化共享资源并在关闭时清理(模型、连接池等) @asynccontextmanager + async def lifespan(app) + FastAPI(lifespan=lifespan) 现代推荐方案,成对逻辑写在一个函数内
阻塞型同步 I/O 的启动/关闭逻辑 lifespan 中相应使用同步代码段即可(由 FastAPI 内部调度) 不需要自行 async def 包装同步阻塞操作
旧项目或历史代码中的 @app.on_event("startup") / @app.on_event("shutdown") 应迁移到 lifespan 已被标记 deprecated 并产生 DeprecationWarning,且与 lifespan 不可混用
挂载(mount)子应用 生命周期逻辑放主应用,子应用事件不会触发 详见 docs/en/docs/advanced/sub-applications.md

若要进一步阅读,可在仓库中对照以下材料:印地语文档 docs/hi/docs/advanced/events.md、英文原版 docs/en/docs/advanced/events.md、三个官方示例 docs_src/events/tutorial003_py310.pydocs_src/events/tutorial001_py310.pydocs_src/events/tutorial002_py310.py,以及 FastAPI 内部实现 fastapi/applications.pyfastapi/routing.py 中关于 lifespan 的处理逻辑和配套测试 tests/test_tutorial/test_events/

登录后查看全文
热门项目推荐
相关项目推荐

项目优选

收起
kernelkernel
deepin linux kernel
C
33
18
ops-transformerops-transformer
本项目是CANN提供的transformer类大模型算子库,实现网络在NPU上加速计算。
C++
1.13 K
2.75 K
pytorchpytorch
作为 Ascend for PyTorch 社区的核心组件,TorchNPU 是昇腾专为 PyTorch 打造的深度学习适配插件,使 PyTorch 框架能够直接调用昇腾 NPU,为开发者提供昇腾 AI 处理器的超强算力。
Python
857
1.35 K
docsdocs
暂无描述
Markdown
897
5.8 K
kernelkernel
openEuler内核是openEuler操作系统的核心,既是系统性能与稳定性的基石,也是连接处理器、设备与服务的桥梁。
C
529
593
ops-nnops-nn
本项目是CANN提供的神经网络类计算算子库,实现网络在NPU上加速计算。
C++
915
1.83 K
jiuwenswarmjiuwenswarm
JiuwenSwarm 是一款基于openJiuwen开发的智能AI Agent,它能够将大语言模型的强大能力,通过你日常使用的各类通讯应用,直接延伸至你的指尖。
Python
3.58 K
1.01 K
ops-mathops-math
本项目是CANN提供的数学类基础计算算子库,实现网络在NPU上加速计算。
C++
1.35 K
1.46 K
cann-learning-hubcann-learning-hub
CANN 学习中心仓,支持在线互动运行、边学边练,提供教程、示例与优化方案,一站式助力昇腾开发者快速上手。
Jupyter Notebook
1.01 K
515
AscendNPU-IRAscendNPU-IR
AscendNPU-IR是基于MLIR(Multi-Level Intermediate Representation)构建的,面向昇腾亲和算子编译时使用的中间表示,提供昇腾完备表达能力,通过编译优化提升昇腾AI处理器计算效率,支持通过生态框架使能昇腾AI处理器与深度调优
C++
547
388