首页
/ FastAPI Lifespan Events:用 lifespan 上下文管理器管理应用启动与关闭的共享资源

FastAPI Lifespan Events:用 lifespan 上下文管理器管理应用启动与关闭的共享资源

2026-09-06 13:31:40作者:郦嵘贵Just

本文围绕 FastAPI 的 Lifespan Events 机制展开:如何在应用启动前关闭后各执行一次逻辑,管理数据库连接池、机器学习模型等被多个请求共享的资源。读完之后,你将掌握推荐的 lifespan 异步上下文管理器写法的完整实现细节、已被弃用的 on_event("startup"/"shutdown") 用法及其互斥规则,并能在源码层面理解这些事件是如何在 FastAPI 应用中落地、在测试中如何被触发的。

核心概念:覆盖整个应用生命周期的钩子

你可以定义在应用启动之前执行的逻辑。这段代码只执行一次,并且发生在应用开始接收请求之前

同样的方式,你也可以定义在应用关闭(shutting down)时执行的逻辑:这段代码同样只执行一次,发生在可能已经处理过大量请求之后。

正因为这段代码在应用开始接收请求之前运行、又在应用结束处理请求之后立刻运行,它覆盖了应用的全部生命周期(Lifespan)——"lifespan" 这个词稍后就会派上用场。

这套机制非常适合初始化那些整个应用都需要用到、且在多个请求间共享的资源,以及事后的清理工作。典型例子:

  • 数据库连接池的创建与释放;
  • 加载一个被所有请求共享的机器学习模型,并在退出时释放内存或 GPU。

使用场景:共享且加载缓慢的 ML 模型

先看一个具体的使用场景,再看如何用上述机制解决它。

假设你有一些机器学习模型,想用来处理请求。🤖

这些模型在所有请求之间共享——不是每个请求一个模型,也不是每个用户一个。再假设加载模型需要相当长的时间,因为它要从磁盘读取大量数据,所以你当然不想在每个请求里都加载一次。

一种偷懒的做法是在模块/文件顶层加载模型,但这意味着即使用户只是在运行一个简单的自动化测试,也必须先把模型加载完——测试会变得很慢,因为它要等模型加载完成后才能执行与被模型无关的的部分代码。

正确的做法是:在请求被处理之前加载模型,但只发生在应用即将开始接收请求的时刻,而不是代码被 import 的时候。

推荐方案:lifespan 参数与异步上下文管理器

你可以利用 FastAPI 应用的 lifespan 参数加一个"上下文管理器"(context manager,下文会解释)来定义这套 startupshutdown 逻辑。

先给出完整示例,再逐段拆解。我们创建一个带 yield 的异步函数 lifespan()

@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()

(完整文件见 docs_src/events/tutorial003_py310.py

这里通过把(模拟的)模型函数放入 ml_models 字典,来模拟昂贵的"加载模型"这一 startup 操作,该操作发生在 yield 之前——这段代码会在应用开始接收请求之前、也就是 startup 阶段执行。

而紧跟在 yield 之后的代码负责卸载模型:它会在应用结束处理请求之后、也就是 shutdown 即将发生之前执行,比如借此释放内存或 GPU 等资源。

提示shutdown 发生在你停止应用的时候——你可能要发布新版本,或者单纯是不想继续运行它了。🤷

Lifespan 函数:yield 前后的两段逻辑

首先注意,我们定义的是一个带 yield 的异步函数。这和带 yield 的依赖(Dependency)非常相似:

@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()

函数的前半部分(yield 之前)在应用启动之前执行;后半部分(yield 之后)在应用结束之后执行。

异步上下文管理器:@asynccontextmanager

可以看到这个函数被 @asynccontextmanager 装饰(来自标准库 contextlib)。

这个装饰器把普通异步函数转换成一个"异步上下文管理器"(async context manager):

from contextlib import asynccontextmanager

@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()

Python 中的上下文管理器是可以用在 with 语句里的对象,例如 open() 就是:

with open("file.txt") as file:
    file.read()

较新版本的 Python 还提供了异步上下文管理器,配合 async with 使用:

async with lifespan(app):
    await do_stuff()

像上面这样创建的上下文管理器或异步上下文管理器,其行为是:进入 with 块之前执行 yield 之前的代码,离开 with 块之后执行 yield 之后的代码。

在我们的示例中并没有直接使用它,而是把它传给 FastAPI 由框架来使用:FastAPI 应用的 lifespan 参数接收一个异步上下文管理器,于是我们可以把新创建的 lifespan 传进去:

app = FastAPI(lifespan=lifespan)

完整的可运行示例(docs_src/events/tutorial003_py310.py)如下,/predict 端点直接读取 startup 阶段加载好的模型字典:

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}

源码层面的实现:lifespan 如何接入 FastAPI

从源码结构看,lifespanFastAPI 构造函数中是一个一等参数。在 fastapi/applications.py 中,其类型注解为 Lifespan[AppType] | None,文档字符串明确写道:"A Lifespan context manager handler. This replaces startup and shutdown functions with a single context manager."

接着,这个参数被原样透传给内部的路由器:fastapi/applications.pyrouting.APIRouter(...) 的构造参数里包含 lifespan=lifespan,与同样被透传的 on_startupon_shutdown 并列——也就是说,真正执行 ASGI Lifespan 协议交互的是底层路由器与 Starlette 的运行时。

替代方案:on_event 事件处理器(已弃用)

警告:处理 startupshutdown 的推荐方式是上面描述的 lifespan 参数。如果你提供了 lifespan 参数,startupshutdown 事件处理器将不再被调用。是全部 lifespan,还是全部旧式 events,二者只能选其一,不能混用。

这一节你大概率可以跳过,以下仅用于理解旧代码或做迁移对照。

lifespan 外,还有一种替代方式定义 startupshutdown 阶段执行的逻辑:声明若干事件处理器(函数),分别在应用启动前或关闭时执行。这些函数可以用 async def 或普通 def 声明。

startup 事件

要添加一个在应用启动前执行的函数,用事件 "startup" 声明它:

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

(完整文件见 docs_src/events/tutorial001_py310.py

在这个例子里,startup 事件处理器函数会向 items 的"数据库"(其实就是一个 dict)初始化几个值。你可以添加多个事件处理器函数;在所有 startup 事件处理器都完成之前,应用不会开始接收请求。

shutdown 事件

要添加一个在应用关闭时执行的函数,用事件 "shutdown" 声明它:

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

(完整文件见 docs_src/events/tutorial002_py310.py

这里 shutdown 事件处理器函数会向文件 log.txt 写入一行文本 "Application shutdown"

说明open()mode="a" 表示 "append"(追加),因此新行会添加在文件已有内容之后,不会覆盖原有内容。

提示:注意这里用的是标准 Python 的 open() 函数来操作文件,涉及需要"等待"磁盘写入的 I/O(输入/输出)操作,但它本身并不使用 async/await。因此事件处理器函数声明为普通 def 而不是 async def

startupshutdown 的联动难题

你的 startupshutdown 逻辑很可能需要互相联动:启动某物然后在关闭时结束它、获取一个资源然后释放它,等等。用两个彼此不共享逻辑或变量的独立函数来做这件事会更困难——你可能需要把值存进全局变量或使用类似的技巧。

正是出于这个原因,官方现在推荐使用上文介绍的 lifespan 方案:yield 前后的两段代码天然共享同一个函数作用域,资源对象可以直接以局部变量传递。

从源码可以印证弃用的态度:fastapi/applications.pyon_event 方法被 @deprecated 装饰器包裹,提示信息为 "on_event is deprecated, use lifespan event handlers instead.";同样,构造参数 on_startupon_shutdown 的文档字符串也写着 "You should instead use the lifespan handlers."(见 fastapi/applications.py)。

技术细节:ASGI Lifespan 协议

给好奇的技术控留一个细节。🤓

在 ASGI 技术规范中,这套机制属于 Lifespan Protocol(生命周期协议) 的一部分,协议中定义了名为 startupshutdown 的事件。FastAPI 的 lifespan 参数与 on_startup/on_shutdown 参数最终都由 Starlette 应用实现来与 ASGI 服务器交换这些协议消息。Starlette 的文档还进一步介绍了 lifespan 状态(lifespan state)的用法——这些状态可以在应用代码的其他部分被读取,例如通过 app.state 存取 startup 阶段初始化的对象。

测试中的触发方式:TestClient 作为上下文管理器

lifespan 逻辑只在应用真正进入"运行"状态时触发,这一点在测试中体现得很直接。以 tests/test_tutorial/test_sql_databases/test_tutorial001.py 为例:

with TestClient(mod_any.app) as c:
    yield c
# Clean up connection explicitly to avoid resource warning
mod_any.engine.dispose()

只有把 TestClient(app) 用在 with 语句中,客户端"进入上下文"时才模拟应用启动(触发 lifespanyield 前的代码),"退出上下文"时模拟应用关闭(触发 yield 后的代码)。若只是简单调用 TestClient(app) 而不用 with,启动/关闭逻辑不会被执行。该测试文件中还有一条值得注意的 TODO 注释:"remove when updating SQL tutorial to use new lifespan API"(tests/test_tutorial/test_sql_databases/test_tutorial001.py),印证了仓库内旧示例仍在使用 on_event 旧式 API、正逐步向 lifespan 迁移的事实。

注意:子应用(Sub Applications / Mounts)不触发主应用的 Lifespan Events

🚨 务必记住:这些 lifespan 事件(startup 和 shutdown)只针对主应用执行,不会为通过 Mount 挂载的子应用(Sub Applications)执行。也就是说,如果你把一个独立的 ASGI 应用挂载到主应用的某个路径下,主应用的 lifespan 启动/关闭逻辑不会替你处理子应用内部的资源;子应用自身若需要 startup/shutdown 逻辑,需要在其自己的应用实例上单独定义。

小结

机制 用法 状态
lifespan 参数 + @asynccontextmanager 函数 app = FastAPI(lifespan=lifespan)yield 前启动、后清理 推荐
@app.on_event("startup"/"shutdown") 装饰独立的 async def / def 函数 已弃用(deprecated)
FastAPI(on_startup=[...], on_shutdown=[...]) 构造参数传入函数列表 已弃用(deprecated)

选择原则:新代码一律使用 lifespan;它既覆盖了 startupshutdown 两个时点,又能让两段逻辑共享同一作用域,避免全局变量;而且一旦提供 lifespan 参数,旧式事件处理器将不再被调用。配合 TestClientwith 用法,还可以在测试中确定性地触发并验证这套生命周期逻辑。

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