什么是lifespan

在了解什么是lifespan之前，我们回顾一下在fastapi框架中，通常我们的处理应用生命周期的方式有哪些？

startup 和 shutdown

在fastapi中提供一个应用生命周期管理事件通知的方式，再之前我们所接触关于管理应用生命周期事件回调通知方式主要有下面方式：

from fastapi import FastAPI

app = FastAPI()

items = {}


@app.on_event("startup")
async def startup_event():
   pass
   print(应用启动)


@app.on_event("shutdown")
async def shutdown_event():
  pass
   print(应用退出时)

在上面的代码中：

@app.on_event("startup")装饰器用于定义启动事件处理函数。这个函数将在应用程序启动时执行，它通常可以用于进行一些常规插件或其他参数等初始化操作，例如建立数据库连接、加载配置文件、设置全局变量等。
对于@app.on_event("shutdown")装饰器则可以用于定义关闭事件处理函数。这个函数将在应用程序关闭时执行，它通常用于处理一些前前我们初始化的插件等释放资源、保存数据或执行清理操作等等。

然后处理上面的相关启动和关闭是处于一个割裂的方式来定义，也就说，如何涉及到一些整体的上下文操作的时候，我们需要依赖一些全局定义或单列方式来处理。

比如假设：我们在startup里面我有一个数据库连接池的对象需要初始化，但是在shutdown结束的时候，我需要进行关闭，但是我额外的方式保持startup里面初始化的数据库连接池的对象的引用，这样才可以在shutdown里面进行调用。

所以就存在割裂的情况。官方对此也有所描述。所以在之前的框架中，我都是依赖使用startup和shutdown。虽然通过封装为类的方式也可以解决类似上述的问题，不过既然官方也再不久将来要废弃这种模式，所以我们就开始学习新的处理方式。那就是：lifespan。

lifespan 新生命周期管理方式

lifespan作用其实和前面提供的startup 和 shutdown一样，都是为了用于解决应用相关生命周期的管理。如下代码示例所示：

from contextlib import asynccontextmanager

from fastapi import FastAPI


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Load the ML model
    print("应用启动-加载模型")
    yield
    print("应用结束-释放模型资源")

app = FastAPI(lifespan=lifespan)


@app.get("/predict")
async def predict(x: float):
    return {"result": "OK"}


if __name__ == "__main__":
    import uvicorn
    import inspect
    # 根据文件路径返回模块名
    # print("app_modeel_name：",inspect.getmodulename(Path(__file__).name))
    # 使用uvicorn.run函数运行了一个应用程序。它指定了应用程序的主机和端口，并且设置了reload参数为True。
    uvicorn.run(app, host='127.0.0.1', port=31112)

启动项目和关闭项目的时候，控制台输出的情况如下：

INFO:     Started server process [21964]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:31112 (Press CTRL+C to quit)
应用启动-加载模型
INFO:     127.0.0.1:54398 - "GET /docs HTTP/1.1" 200 OK
INFO:     127.0.0.1:54398 - "GET /openapi.json HTTP/1.1" 200 OK
INFO:     Shutting down
应用结束-释放模型资源
INFO:     Waiting for application shutdown.
INFO:     Application shutdown complete.
INFO:     Finished server process [21964]

对比：


@app.on_event("startup")
async def startup_event():
    print("startup-应用启动-加载模型")

@app.on_event("shutdown")
async def startup_event():
    print("shutdown-应用结束-释放模型资源")

得到的效果是一样。

INFO:     Started server process [30300]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:31112 (Press CTRL+C to quit)
startup-应用启动-加载模型
INFO:     Shutting down
INFO:     Waiting for application shutdown.
INFO:     Application shutdown complete.
INFO:     Finished server process [30300]
shutdown-应用结束-释放模型资源

lifespan 的局限和第三方库

假如我有非常多的插件需要实现类似上面这种基于lifespan方式的生命周期的管理，那我无法向使用@app.on_event("startup")和@app.on_event("shutdown")进行多次的重复的注册和分类。

为此我们可以引入其他第三方库：

https://github.com/uriyyo/fastapi-lifespan-manager

基于该库我们可以很好进行分离多个注册的实现，如提供的示例代码所示：

from typing import AsyncIterator

from fastapi import FastAPI
from redis.asyncio import Redis
from sqlalchemy.ext.asyncio import create_async_engine

from fastapi_lifespan_manager import LifespanManager, State

manager = LifespanManager()


@manager.add
async def setup_db(app: FastAPI) -> AsyncIterator[State]:
    engine = await create_async_engine("postgresql+asyncpg://user:password@localhost/db")

    yield {"db": engine}

    await engine.dispose()


@manager.add
async def setup_cache(app: FastAPI) -> AsyncIterator[State]:
    redis = await Redis.from_url("redis://localhost:6379/0")

    yield {"cache": redis}

    await redis.close()
    await redis.wait_closed()


app = FastAPI(lifespan=manager)

fastapi-lifespan-manager原理实现简单分析

fastapi-lifespan-manager的实现也比较简单，我们首先贴出具体的代码，如下：

from __future__ import annotations

import inspect
from contextlib import AsyncExitStack, asynccontextmanager, contextmanager
from dataclasses import dataclass
from typing import (
    TYPE_CHECKING,
    Any,
    AsyncContextManager,
    AsyncIterator,
    Callable,
    ContextManager,
    Dict,
    Generic,
    Iterable,
    List,
    Literal,
    Optional,
    Union,
    cast,
    overload,
)

from fastapi import FastAPI
from fastapi.concurrency import contextmanager_in_threadpool
from typing_extensions import Self

from .types import AnyContextManager, AnyState, RawLifespan, State, TApp


def _maybe_call(call: Callable[..., AnyContextManager], app: Optional[TApp]) -> AnyContextManager:
    try:
        return call(app)
    except TypeError:
        return call()


def _convert_raw_lifespan_to_ctx(
    app: TApp,
    raw_lifespan: RawLifespan[TApp],
) -> Union[ContextManager[AnyState], AsyncContextManager[AnyState]]:
    if inspect.isasyncgenfunction(raw_lifespan):
        return _maybe_call(asynccontextmanager(raw_lifespan), app)
    if inspect.isgeneratorfunction(raw_lifespan):
        return _maybe_call(contextmanager(raw_lifespan), app)

    return _maybe_call(cast(Callable[..., AnyContextManager], raw_lifespan), app)


@asynccontextmanager
async def _run_raw_lifespan(raw_lifespan: RawLifespan[TApp], app: TApp) -> AsyncIterator[AnyState]:
    ctx = _convert_raw_lifespan_to_ctx(app, raw_lifespan)

    actx: AsyncContextManager[AnyState]
    actx = contextmanager_in_threadpool(ctx) if isinstance(ctx, ContextManager) else ctx

    async with actx as state:
        yield state


class LifespanManager(Generic[TApp]):
    lifespans: List[RawLifespan[TApp]]

    if TYPE_CHECKING:

        @overload
        def __new__(cls, __lifespans: Iterable[RawLifespan[TApp]], /) -> LifespanManager[TApp]:
            pass

        @overload
        def __new__(cls, __lifespans: Literal[None] = None, /) -> LifespanManager[FastAPI]:
            pass

        def __new__(cls, __lifespans: Optional[Iterable[RawLifespan[TApp]]] = ..., /) -> LifespanManager[Any]:
            pass

    else:

        def __init__(self, lifespans: Optional[Iterable[RawLifespan[TApp]]] = None, /) -> None:
            self.lifespans = [*(lifespans or [])]

    def add(self, lifespan: RawLifespan[TApp]) -> RawLifespan[TApp]:
        self.lifespans.append(lifespan)
        return lifespan

    def remove(self, lifespan: RawLifespan[TApp]) -> None:
        self.lifespans.remove(lifespan)

    def include(self, other: Self) -> None:
        self.lifespans.extend(other.lifespans)

    @asynccontextmanager
    async def __call__(self, app: TApp) -> AsyncIterator[State]:
        async with AsyncExitStack() as astack:
            state: Dict[str, Any] = {}

            for raw_lifespan in self.lifespans:
                sub_state = await astack.enter_async_context(_run_raw_lifespan(raw_lifespan, app))

                if sub_state:
                    state.update(sub_state)

            yield state


if not TYPE_CHECKING:
    # mypy complains about using a dataclass directly with LifeSpanManager
    LifeSpanManager = dataclass(init=False)(LifespanManager)

__all__ = [
    "LifespanManager",
]

具体代码分析如下：

首先class LifespanManager(Generic[TApp])，这个类的定义，它表示的是我们定义了一个泛型类 "LifespanManager"，而其中 "[TApp]" 则表示该类是一个泛型类，它呐可以接受一个类型参数 TApp。也就是说这个类型参数可以在创建类的实例时被指定，用于表示应用程序的类型。在这里通过定义为一个泛型的类，可以很好的使得这个类可以处理不同类型的应用程序。
其次在类中我们定义了一个"lifespans" 的列表属性，它主要用于存储我们所有的需要处理的原始生命周期对象。对于这个相关加入的管理对象也提供相关如 "add"、"remove" 和 "include"，用于添加、删除和合并生命周期对象的方法。
接着是关于 if TYPE_CHECKING等类型检测器的检测，其中if TYPE_CHECKING:：表示的意思是当静态类型检查器运行时，则执行内部相关的代码逻辑。内部的逻辑代码则定义了三个重载方法 __new__，它们主要是用于创建 LifespanManager 类的实例。这些重载方法接受不同的参数类型，并返回对应的泛型类型 LifespanManager[TApp] 的实例。具体来说：其中内部的代码逻辑说明如下：

相关说明如下

第一个重载方法接受一个可迭代对象 __lifespans，类型为 Iterable[RawLifespan[TApp]]，并返回 LifespanManager[TApp] 类型的实例。

第二个重载方法没有参数，其默认值为 None，类型为 Literal[None]，并返回 LifespanManager[FastAPI] 类型的实例。

第三个重载方法接受一个可选的可迭代对象 __lifespans，类型为 Optional[Iterable[RawLifespan[TApp]]]，并返回 LifespanManager[Any] 类型的实例。

否则，如果当静态类型检查器未运行时，则执行另一个初始化 LifespanManager 类的实例的逻辑处理。其中改逻辑中的方法接受一个可选的可迭代对象 lifespans，类型是一个 Optional[Iterable[RawLifespan[TApp]]]，并将其赋值给实例的 lifespans 属性。如果 lifespans 参数为 None，则将一个空列表赋值给 lifespans 属性。

最关键的核心处理多个生命周期对象管理在如下代码中：

@asynccontextmanager
    async def __call__(self, app: TApp) -> AsyncIterator[State]:
        async with AsyncExitStack() as astack:
            state: Dict[str, Any] = {}

            for raw_lifespan in self.lifespans:
                sub_state = await astack.enter_async_context(_run_raw_lifespan(raw_lifespan, app))

                if sub_state:
                    state.update(sub_state)

            yield state

首先在类的内部中，这段代码声明

定义了一个异步上下文管理器 __call__方法，在这个上下文管理器中，它可以传入一个泛型参数 TApp，并返回一个异步迭代器 AsyncIterator[State]类型。
紧接着，然后使用 async with AsyncExitStack() as astack: 创建了一个异步上下文管理器 astack。【PS：AsyncExitStack 是一个异步版本的 ExitStack，用于管理一组异步上下文，确保它们都被正确地进入和退出。】
然后在 astack 的上下文里面创建了一个空字典 state: Dict[str, Any] = {}，用于存储状态信息。接着对通过遍历迭代self.lifespans（即 LifespanManager 实例的 lifespans 属性），对每个生命周期的对象相关的操作。内部逻辑相关操作目的是用于在异步上下文管理器中执行一系列生命周期过程，并将每个过程的状态信息收集到一个字典中，然后通过异步迭代器返回该字典。

其中相关的操作细节说明如下：

使用 await astack.enter_async_context(_run_raw_lifespan(raw_lifespan, app)) 进入一个异步上下文环境，并将返回的 sub_state 赋值给 sub_state 变量。
_run_raw_lifespan ，是外部一个定义上下文的管理器对象。它可以是一个函数或方法，主要的用于执行某个生命周期过程，并返回一个表示该过程状态的字典（或可等效于字典的对象）。
app 是传递给 call 方法的参数，用于在生命周期过程中进行操作。如果 sub_state 存在（即非空），则使用 state.update(sub_state) 更新 state 字典，将 sub_state 的内容合并到 state 中。
最后，通过 yield state 将 state 字典作为异步迭代器的值返回出去。这也就意味着在使用这个异步上下文管理器时，可以通过 async for 循环来迭代并获取每个生命周期过程的状态信息。

_run_raw_lifespan外部的上下文管理器

接着我们分析一下外部的_run_raw_lifespan，它代码如下所示：

@asynccontextmanager
async def _run_raw_lifespan(raw_lifespan: RawLifespan[TApp], app: TApp) -> AsyncIterator[AnyState]:
    ctx = _convert_raw_lifespan_to_ctx(app, raw_lifespan)

    actx: AsyncContextManager[AnyState]
    actx = contextmanager_in_threadpool(ctx) if isinstance(ctx, ContextManager) else ctx

    async with actx as state:
        yield state

上面的代码简单分析如下：

首先定义了一个异步上下文管理器 _run_raw_lifespan，它是一个函数，它接受两个参数：raw_lifespan 和 app，并返回一个异步迭代器 AsyncIterator[AnyState]。
然后通过 _convert_raw_lifespan_to_ctx(app, raw_lifespan) 将 raw_lifespan 转换为一个上下文对象 ctx，该对象可以在上下文环境中被进入和退出。

接下下来就是根据 ctx 的类型进行不同机制的处理：

如果 ctx 是一个上下文管理器（也就是说如它这个上下文它实现了 ContextManager 接口），则使用 contextmanager_in_threadpool(ctx) 将其转换为一个在线程池中运行的上下文管理器，并将结果赋值给 actx。
如果 ctx 自身已经是一个异步上下文管理器了，那么久直接将其赋值给 actx。
然后在 actx 的上下文中，使用 async with actx as state: 进入异步上下文环境，并将返回的 state 赋值给 state 变量。执行到了最后，则通过 yield state 将 state 作为异步迭代器的值返回。

_convert_raw_lifespan_to_ctx转换上下文

在进行把迭代器对象转化为一个上下文对象 ctx中，我们涉及到了_convert_raw_lifespan_to_ctx方法，它的代码如下：

def _convert_raw_lifespan_to_ctx(
    app: TApp,
    raw_lifespan: RawLifespan[TApp],
) -> Union[ContextManager[AnyState], AsyncContextManager[AnyState]]:
    if inspect.isasyncgenfunction(raw_lifespan):
        return _maybe_call(asynccontextmanager(raw_lifespan), app)
    if inspect.isgeneratorfunction(raw_lifespan):
        return _maybe_call(contextmanager(raw_lifespan), app)

    return _maybe_call(cast(Callable[..., AnyContextManager], raw_lifespan), app)

内部代码主要是对传入的类型对象的检测，根据不同的类型返回不同的对象，如下说明：

使用inspect.isasyncgenfunction(raw_lifespan) 检查 raw_lifespan ，是否是一个异步生成器函数（async generator function）。如果是异步生成器函数，则通过 asynccontextmanager 将其转换为一个异步上下文管理器对象，并使用 _maybe_call 函数调用该对象，并将 app 作为参数传递给它。
使用inspect.isgeneratorfunction(raw_lifespan) 检查 raw_lifespan 是否是生成器函数（generator function）。如果是生成器函数，则通过 contextmanager 将其转换为一个上下文管理器对象，并使用 _maybe_call 函数调用该对象，并将 app 作为参数传递给它。
最终如果 raw_lifespan 不是异步生成器函数或生成器函数，那么它应该是一个可调用对象。代码使用 cast(Callable[..., AnyContextManager], raw_lifespan) 将 raw_lifespan 强制转换为一个可调用对象，并将返回的对象视为上下文管理器对象。然后，使用 _maybe_call 函数调用该对象，并将 app 作为参数传递给它。

最终执行到最后，则根据转换得到的上下文管理器对象的类型对象，函数会根据实际集结果返回一个适当的对象。

如果是异步上下文管理器，则返回 AsyncContextManager[AnyState] 类型的对象；
如果是普通上下文管理器，则返回 ContextManager[AnyState] 类型的对象。

如此，引入该库，我们也可以实现类似多个生命周期对象的管理。

文笔有限，如有笔误或错误！欢迎批评指正！感谢各位大佬！有什么问题也可以随时交流！

结尾

END

简书：https://www.jianshu.com/u/d6960089b087

掘金：https://juejin.cn/user/2963939079225608

公众号：微信搜【程序员小钟同学】

新开QQ群号，欢迎随时加群交流，相互学习。QQ群号：247491107

小钟同学 | 文【欢迎一起学习交流】| QQ：308711822

ITPub博客

聊聊关于fastapi lifespan 生命周期事件