生命周期事件¶
您可以定义在应用程序启动之前应该执行的逻辑（代码）。这意味着这些代码将一次性在应用程序开始接收请求之前执行。
同样，您可以定义在应用程序关闭时应该执行的逻辑（代码）。在这种情况下，这些代码将在处理了许多请求之后一次性执行。
因为这些代码是在应用程序开始处理请求之前执行，并在处理完请求之后立即执行，所以它涵盖了整个应用程序的生命周期（“生命周期”这个词一会儿就会变得很重要 😉）。
这对于设置您需要在整个应用程序中使用的资源非常有用，这些资源可以在请求之间共享，并且/或者您需要在之后清理。例如，数据库连接池，或加载共享的机器学习模型。
用例¶
让我们从一个用例开始，然后看看如何用它来解决。
假设您有一些机器学习模型要用来处理请求。 🤖
相同的模型在请求之间共享，所以不是每个请求一个模型，或每个用户一个模型之类的。
假设加载模型可能需要相当长的时间，因为它需要从磁盘读取大量数据。所以您不想为每个请求都这样做。
您可以将其加载到模块/文件的顶层，但这也意味着即使您只是运行一个简单的自动化测试，它也会加载模型，然后该测试将变慢，因为它必须等待模型加载才能运行代码的独立部分。
这就是我们将要解决的问题：在处理请求之前加载模型，但仅在应用程序开始接收请求之前，而不是在代码加载时。
生命周期¶
您可以使用 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 等资源。
提示
当您停止应用程序时，将发生 shutdown。
也许您需要启动一个新版本，或者您只是厌倦了运行它。 🤷
生命周期函数¶
首先要注意的是，我们正在定义一个带有 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，要么是事件，而不是两者都用。
您可能可以跳过这一部分。
还有一个备用的方法来定义在启动和关闭期间执行的逻辑。
您可以定义需要在应用程序启动前执行，或在应用程序关闭时执行的事件处理程序（函数）。
这些函数可以用 async def 或标准的 def 来声明。
startup 事件¶
要添加一个在应用程序启动前运行的函数，请使用 "startup" 事件声明它。
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 事件处理程序函数将使用一些值初始化“数据库”（只是一个 dict）项。
您可以添加多个事件处理程序函数。
在所有 startup 事件处理程序完成之前，您的应用程序将不会开始接收请求。
shutdown 事件¶
要添加一个在应用程序关闭时运行的函数，请使用 "shutdown" 事件声明它。
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 事件处理程序函数会将文本行 "Application shutdown" 写入到文件 log.txt 中。
信息
在 open() 函数中，mode="a" 表示“追加”，所以，该行将被添加到文件内容的末尾，而不会覆盖之前的内容。
提示
请注意，在这种情况下，我们使用的是标准的 Python open() 函数，它与文件交互。
因此，它涉及 I/O（输入/输出），这需要“等待”将内容写入磁盘。
但是 open() 不使用 async 和 await。
所以，我们使用标准的 def 来声明事件处理程序函数，而不是 async def。
startup 和 shutdown 一起使用¶
很有可能您的启动和关闭逻辑是关联的，您可能想启动某项然后结束它，获取一个资源然后释放它，等等。
在不共享逻辑或变量的独立函数中执行此操作更加困难，因为您需要将值存储在全局变量或类似的技巧中。
因此，现在建议改用上面解释的 lifespan。
技术细节¶
只是一个给好奇的极客的技术细节。 🤓
底层来说，在 ASGI 技术规范中，这是 Lifespan Protocol 的一部分，它定义了称为 startup 和 shutdown 的事件。
子应用程序¶
🚨 请记住，这些生命周期事件（启动和关闭）仅为应用程序本身执行，而不是为 子应用程序 - 挂载 执行。