生命周期事件¶

你可以在应用启动前定义需要执行的逻辑（代码）。这意味着该代码将在应用开始接收请求之前执行一次。

同样地，你可以定义在应用关闭时需要执行的逻辑（代码）。在这种情况下，该代码将在处理完可能多个请求后执行一次。

由于这段代码在应用开始接收请求之前执行，并在结束处理请求之后立即执行，它覆盖了整个应用的生命周期（"lifespan"这个词马上会很重要😉）。

这对于设置需要在整个应用中使用的资源（这些资源在请求间共享）和/或之后需要清理的资源非常有用。例如，数据库连接池或加载共享的机器学习模型。

用例¶

让我们从一个示例用例开始，然后看看如何用这个功能解决它。

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

相同的模型在请求间共享，因此不是每个请求一个模型，也不是每个用户一个模型或类似的情况。

假设加载模型可能需要相当长的时间，因为它必须从磁盘读取大量数据。所以你不想为每个请求都加载它。

你可以在模块/文件的顶层加载它，但这意味着即使你只是运行一个简单的自动化测试，它也会加载模型，那么该测试会变得缓慢，因为它必须等待模型加载完成后才能运行代码的独立部分。

这就是我们要解决的问题，让我们在请求被处理之前加载模型，但只在应用开始接收请求之前加载，而不是在代码加载时加载。

生命周期¶

你可以使用 FastAPI 应用的 lifespan 参数和一个"上下文管理器"（我马上会告诉你这是什么）来定义这种启动和关闭逻辑。

让我们从一个例子开始，然后详细查看。

我们创建一个带有 yield 的异步函数 lifespan()，如下所示：

Python 3.8+

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 等资源。

Tip

shutdown 会在你停止应用时发生。

也许你需要启动一个新版本，或者你只是厌倦了运行它。🤷

生命周期函数¶

首先要注意的是，我们正在定义一个带有 yield 的异步函数。这与带有 yield 的依赖项非常相似。

Python 3.8+

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 装饰。

这将该函数转换为所谓的"异步上下文管理器"。

Python 3.8+

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 异步上下文管理器传递给它。

Python 3.8+

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}

替代事件（已弃用）¶

Warning

处理启动和关闭的推荐方式是使用如上所述的 FastAPI 应用的 lifespan 参数。如果你提供了 lifespan 参数，startup 和 shutdown 事件处理程序将不再被调用。要么全部使用 lifespan，要么全部使用事件，不能两者都用。

你可能会跳过这部分。

还有一种替代方法来定义在启动和关闭期间要执行的逻辑。

你可以定义需要在应用启动之前或应用关闭时执行的事件处理程序（函数）。

这些函数可以用 async def 或普通的 def 声明。

`startup` 事件¶

要添加一个应在应用启动前运行的函数，请使用事件 "startup" 声明它：

Python 3.8+

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" 声明它：

Python 3.8+

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。

Info

在 open() 函数中，mode="a" 表示"追加"，因此，该行将在该文件中的任何内容之后添加，而不会覆盖先前的内容。

Tip

请注意，在这种情况下，我们使用的是与文件交互的标准 Python open() 函数。

因此，它涉及 I/O（输入/输出），需要"等待"内容写入磁盘。

但是 open() 不使用 async 和 await。

因此，我们使用标准的 def 而不是 async def 来声明事件处理函数。

`startup` 和 `shutdown` 一起¶

你的启动和关闭逻辑很可能是相连的，你可能想启动某些东西然后结束它，获取资源然后释放它，等等。

在不相连的函数中这样做，不共享逻辑或变量，会更加困难，因为你需要在全局变量中存储值或使用类似的技巧。

因此，现在建议改用上面解释的 lifespan。

技术细节¶

只是给好奇的极客们的一个技术细节。🤓

在底层，在 ASGI 技术规范中，这是 Lifespan Protocol 的一部分，它定义了称为 startup 和 shutdown 的事件。

Info

你可以在 Starlette 的生命周期文档中阅读更多关于 Starlette lifespan 处理程序的信息。

包括如何处理可以在代码其他区域使用的生命周期状态。

子应用¶

🚨 请记住，这些生命周期事件（启动和关闭）将仅对主应用程序执行，不对子应用 - 挂载执行。

生命周期事件¶

用例¶

生命周期¶

生命周期函数¶

异步上下文管理器¶

替代事件（已弃用）¶

startup 事件¶

shutdown 事件¶

startup 和 shutdown 一起¶

技术细节¶

子应用¶

`startup` 事件¶

`shutdown` 事件¶

`startup` 和 `shutdown` 一起¶