生命周期事件¶
你可以定义在应用启动前执行的逻辑（代码）。这意味着在应用开始接收请求之前，这些代码只会被执行一次。
同样地，你可以定义在应用关闭时应执行的逻辑。在这种情况下，这段代码将在处理可能的多次请求后执行一次。
因为这段代码在应用开始接收请求之前执行，也会在处理可能的若干请求之后执行，它覆盖了整个应用程序的生命周期（"生命周期"这个词很重要😉）。
这对于设置你需要在整个应用中使用的资源非常有用，这些资源在请求之间共享，你可能需要在之后进行释放。例如，数据库连接池，或加载一个共享的机器学习模型。
用例¶
让我们从一个示例用例开始，看看如何解决它。
假设你有几个机器学习的模型，你想要用它们来处理请求。
相同的模型在请求之间是共享的，因此并非每个请求或每个用户各自拥有一个模型。
假设加载模型可能需要相当长的时间，因为它必须从磁盘读取大量数据。因此你不希望每个请求都加载它。
你可以在模块/文件的顶部加载它，但这也意味着即使你只是在运行一个简单的自动化测试，它也会加载模型，这样测试将变慢，因为它必须在能够独立运行代码的其他部分之前等待模型加载完成。
这就是我们要解决的问题——在处理请求前加载模型，但只是在应用开始接收请求前，而不是代码执行时。
生命周期 lifespan¶
你可以使用FastAPI()应用的lifespan参数和一个上下文管理器（稍后我将为你展示）来定义启动和关闭的逻辑。
让我们从一个例子开始，然后详细介绍。
我们使用yield创建了一个异步函数lifespan()像这样：
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"](x)
return {"result": result}
在这里，我们在 yield 之前将（虚拟的）模型函数放入机器学习模型的字典中，以此模拟加载模型的耗时启动操作。这段代码将在应用程序开始处理请求之前执行，即启动期间。
然后，在 yield 之后，我们卸载模型。这段代码将会在应用程序完成处理请求后执行，即在关闭之前。这可以释放诸如内存或 GPU 之类的资源。
提示
关闭事件只会在你停止应用时触发。
可能你需要启动一个新版本，或者你只是你厌倦了运行它。 🤷
生命周期函数¶
首先要注意的是，我们定义了一个带有 yield 的异步函数。这与带有 yield 的依赖项非常相似。
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"](x)
return {"result": result}
这个函数在 yield之前的部分，会在应用启动前执行。
剩下的部分在 yield 之后，会在应用完成后执行。
异步上下文管理器¶
如你所见，这个函数有一个装饰器 @asynccontextmanager 。
它将函数转化为所谓的“异步上下文管理器”。
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"](x)
return {"result": result}
在 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 传给它。
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"](x)
return {"result": result}
替代事件（弃用）¶
警告
配置启动和关闭事件的推荐方法是使用 FastAPI() 应用的 lifespan 参数，如前所示。如果你提供了一个 lifespan 参数，启动（startup）和关闭（shutdown）事件处理器将不再生效。要么使用 lifespan，要么配置所有事件，两者不能共用。
你可以跳过这一部分。
有一种替代方法可以定义在启动和关闭期间执行的逻辑。
FastAPI 支持定义在应用启动前，或应用关闭时执行的事件处理器（函数）。
事件函数既可以声明为异步函数（async def），也可以声明为普通函数（def）。
startup 事件¶
使用 startup 事件声明 app 启动前运行的函数：
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 事件处理器函数为项目数据库（只是字典）提供了一些初始值。
FastAPI 支持多个事件处理器函数。
只有所有 startup 事件处理器运行完毕，FastAPI 应用才开始接收请求。
shutdown 事件¶
使用 shutdown 事件声明 app 关闭时运行的函数：
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" 指的是追加。因此这行文本会添加在文件已有内容之后，不会覆盖之前的内容。
提示
注意，本例使用 Python open() 标准函数与文件交互。
这个函数执行 I/O（输入/输出）操作，需要等待内容写进磁盘。
但 open() 函数不支持使用 async 与 await。
因此，声明事件处理函数要使用 def，不能使用 asnyc def。
startup 和 shutdown 一起使用¶
启动和关闭的逻辑很可能是连接在一起的，你可能希望启动某个东西然后结束它，获取一个资源然后释放它等等。
在不共享逻辑或变量的不同函数中处理这些逻辑比较困难，因为你需要在全局变量中存储值或使用类似的方式。
因此，推荐使用 lifespan 。
技术细节¶
只是为好奇者提供的技术细节。🤓
在底层，这部分是生命周期协议的一部分，参见 ASGI 技术规范，定义了称为启动（startup）和关闭（shutdown）的事件。
子应用¶
🚨 FastAPI 只会触发主应用中的生命周期事件，不包括子应用 - 挂载中的。