生命周期事件¶

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

同样地，你可以定义在应用程序关闭时执行的逻辑（代码）。在这种情况下，这段代码将在处理了可能大量的请求之后，执行一次。

由于这段代码在应用程序开始接收请求之前执行，并在它结束处理请求之后紧接着执行，因此它覆盖了整个应用程序的生命周期（“生命周期/lifespan”这个词稍后会很重要 😉）。

这对于设置整个应用都需要使用、且在请求间共享的资源非常有用，或者用于随后需要清理的资源。例如，数据库连接池，或加载共享的机器学习模型。

应用场景¶

让我们从一个用例开始，看看如何用它来解决问题。

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

这些模型在请求之间是共享的，所以不是每个请求一个模型，也不是每个用户一个模型或类似的结构。

想象一下，加载模型可能需要相当长的时间，因为它必须从磁盘读取大量数据。所以你不想在每个请求中都做这件事。

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

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

生命周期¶

你可以使用 FastAPI 应用的 lifespan 参数和一个“上下文管理器”（稍后我会展示这是什么）来定义这种启动和关闭逻辑。

让我们从一个示例开始，然后再详细说明。

我们像这样创建一个带有 yield 的异步函数 lifespan()

Python 3.10+

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 的依赖项非常相似。

Python 3.10+

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.10+

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.10+

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

Python 3.10+

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.10+

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 的事件。

注意

你可以通过 Starlette 的 Lifespan 文档阅读更多关于 Starlette lifespan 处理器的内容。

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

子应用¶

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

生命周期事件¶

应用场景¶

生命周期¶

生命周期函数¶

异步上下文管理器¶

替代事件（已弃用）¶

startup 事件¶

shutdown 事件¶

startup 和 shutdown 同时使用¶

技术细节¶

子应用¶

`startup` 事件¶

`shutdown` 事件¶

`startup` 和 `shutdown` 同时使用¶