Starlette Lifespan 生命周期管理启动关闭钩子、State 状态共享与类型安全访问实战【免费下载链接】starletteThe little ASGI framework that shines. 项目地址: https://gitcode.com/gh_mirrors/st/starlette导读在 Starlette 这类 ASGI 框架中数据库连接池、HTTP 客户端、缓存等资源的创建与销毁往往需要与应用进程的启动、关闭保持同步。lifespan就是 Starlette 提供的官方机制通过一个异步上下文管理器钩子让你在应用开始接收请求之前完成初始化在所有连接关闭、后台任务结束后完成清理。本文基于 docs/lifespan.md 展开结合当前仓库源码完整讲解 lifespan 的基本用法、生命周期时序保证、state状态共享、属性式与字典式两种访问方式后者在 Starlette 0.52.0 起引入可显著提升类型安全以及在TestClient中如何正确触发 lifespan 的测试写法。认识 lifespan应用的启动与关闭钩子Starlette 应用可以注册一个 lifespan 处理器用来承载应用启动前与应用关闭时需要执行的代码。其写法依托 Python 标准库的contextlib.asynccontextmanager装饰器import contextlib from starlette.applications import Starlette contextlib.asynccontextmanager async def lifespan(app): async with some_async_resource(): print(Run at startup!) yield print(Run on shutdown!) routes [ ... ] app Starlette(routesroutes, lifespanlifespan)要点解读yield之前的代码块在应用启动阶段执行对应打印 Run at startup!yield之后的代码块在应用关闭阶段执行对应打印 Run on shutdown!。通过Starlette(..., lifespanlifespan)把该处理器挂载到应用上函数签名中的app参数即当前应用实例可用于读取app.state等。这种结构天然适配随用随建、随关随毁的资源管理模式例如async with some_async_resource():可以换成任何支持异步上下文协议的资源。生命周期时序的两条硬性保证原文档明确了两条重要的时序语义请求不会被提前处理Starlette 在 lifespan 运行完成之前不会开始处理任何进入的请求。也就是说yield真正放行了请求服务如果启动阶段抛错或未完成应用不会对外提供服务。关闭在一切收尾之后lifespan 的 teardownyield之后的代码会在所有连接都已关闭、所有进程内后台任务都已完成之后才执行。这保证了你关闭连接池时不会再收到新的请求或后台任务占用。对于需要在后台维护异步任务的场景原文档建议优先使用anyio.create_task_group()来管理这些异步任务使其纳入同一生命周期管理范围避免任务游离于应用之外。源码层面的实现印证lifespan参数最终从Starlette.__init__传入内部的Router见 starlette/applications.py 的构造函数与self.router Router(routes, lifespanlifespan)。真正的执行逻辑在 starlette/routing.py 的Router.lifespan方法中async def lifespan(self, scope: Scope, receive: Receive, send: Send) - None: started False app: Any scope.get(app) await receive() # 等待 lifespan.startup 消息 try: async with self.lifespan_context(app) as maybe_state: if maybe_state is not None: if state not in scope: raise RuntimeError(The server does not support state in the lifespan scope.) scope[state].update(maybe_state) await send({type: lifespan.startup.complete}) started True await receive() # 等待 lifespan.shutdown 消息 except BaseException: exc_text traceback.format_exc() if started: await send({type: lifespan.shutdown.failed, message: exc_text}) else: await send({type: lifespan.startup.failed, message: exc_text}) raise else: await send({type: lifespan.shutdown.complete})从这段实现可以推断几个关键事实lifespan 底层基于 ASGI 规范的lifespanscope 协议服务器先发送lifespan.startup应用在启动阶段完成后回送lifespan.startup.complete进程结束时服务器发送lifespan.shutdown应用完成 teardown 后回送lifespan.shutdown.complete。若应用没有显式注册 lifespanRouter.__init__会为其挂上一个空操作的_DefaultLifespan见 starlette/routing.py保证协议流程始终存在、应用无需关心是否有自定义生命周期逻辑。lifespan 处理器抛出的任何异常都会被捕获并转成lifespan.startup.failed/lifespan.shutdown.failed消息上报同时重新抛出方便服务器层记录错误。旧式 lifespan 写法与弃用提示从 starlette/routing.py 的源码可以看出Starlette 目前仍兼容两种旧式写法但会发出StarletteDeprecationWarning弃用告警直接传入异步生成器函数async def lifespan(app): ... yield ...直接传入同步生成器函数def lifespan(app): ... yield ...。两者都会在Router.__init__中被包装成标准上下文管理器但官方建议统一改用contextlib.asynccontextmanager装饰的形式这也是原文档所有示例采用的写法。在 starlette/types.py 中可以看到 lifespan 的类型定义StatelessLifespan Callable[[AppType], AbstractAsyncContextManager[None]] StatefulLifespan Callable[[AppType], AbstractAsyncContextManager[Mapping[str, Any]]] Lifespan StatelessLifespan[AppType] | StatefulLifespan[AppType]即不带状态共享的 lifespan 上下文产出None带状态共享的则产出Mapping[str, Any]通常是字典。Lifespan State在生命周期与请求之间共享对象启动时创建的资源数据库连接池、HTTP 客户端等如果只在 lifespan 内部持有请求处理器将无从获取。为此 Starlette 引入了state概念它本质上是一个字典用于在 lifespan 与各个请求之间共享对象。原文档给出了完整示例——在启动阶段创建一个httpx.AsyncClient并把它注入到请求处理函数中import contextlib from typing import AsyncIterator, TypedDict import httpx from starlette.applications import Starlette from starlette.requests import Request from starlette.responses import PlainTextResponse from starlette.routing import Route class State(TypedDict): http_client: httpx.AsyncClient contextlib.asynccontextmanager async def lifespan(app: Starlette) - AsyncIterator[State]: async with httpx.AsyncClient() as client: yield {http_client: client} async def homepage(request: Request) - PlainTextResponse: client request.state.http_client response await client.get(https://www.example.com) return PlainTextResponse(response.text) app Starlette( lifespanlifespan, routes[Route(/, homepage)] )这里值得注意的细节lifespan 通过yield {http_client: client}把状态以字典形式吐给应用只要资源存活在async with块内整个应用运行期间都能使用该客户端。请求端通过request.state.http_client属性式访问取回对象与 lifespan 内共享的是同一个httpx.AsyncClient实例因此连接复用、并发安全等行为完全一致。原文档特别强调请求端收到的state是 lifespan 处理器中state的浅拷贝shallow copy。即字典本身是复制出来的但字典中的值如client对象仍是同一个引用——这正是共享对象得以成立的原因。源码印证state 的流转链路结合源码可以还原 state 的完整流转路径lifespan 中yield出的字典在 starlette/routing.py 中被合并进 ASGI scope 的state字段scope[state].update(maybe_state)。当请求到达时Request.state属性见 starlette/requests.py会基于同一个scope[state]字典构造State对象self._state State(self.scope[state])从而让请求侧读到与 lifespan 相同的内容。State类的实现位于 starlette/datastructures.py它把内部数据存放在self._state字典中并同时实现了属性式访问__getattr__/__setattr__与字典式访问__getitem__/__setitem__两套协议这也正是下一节两种访问语法都能工作的底层原因。访问 State属性式与字典式两种语法state既可以用属性式语法访问request.state.foo也可以用字典式语法访问request.state[foo]。字典式语法是在Starlette 0.52.02026 年 1 月引入的其初衷是随着Request变成对 state 类型参数化generic的类型字典式访问可以带来更好的类型安全。原文档给出了使用TypedDict配合泛型Request[State]的完整示例from collections.abc import AsyncIterator from contextlib import asynccontextmanager from typing import TypedDict import httpx from starlette.applications import Starlette from starlette.requests import Request from starlette.responses import PlainTextResponse from starlette.routing import Route class State(TypedDict): http_client: httpx.AsyncClient asynccontextmanager async def lifespan(app: Starlette) - AsyncIterator[State]: async with httpx.AsyncClient() as client: yield {http_client: client} async def homepage(request: Request[State]) - PlainTextResponse: client request.state[http_client] reveal_type(client) # Revealed type is httpx.AsyncClient response await client.get(https://www.example.com) return PlainTextResponse(response.text) app Starlette(lifespanlifespan, routes[Route(/, homepage)])关键差异在于类型标注request: Request[State]把请求的类型参数声明为State这个TypedDict随后request.state[http_client]会被类型检查器精确推断为httpx.AsyncClient示例中的reveal_type(client)表明推导类型正是httpx.AsyncClient从而在编译期或编辑器内就拦截键名拼写错误、取值类型不符等问题。这种写法同样适用于 WebSocket 端点WebSocket同样是泛型化的async def websocket_endpoint(websocket: WebSocket[State]) - None: await websocket.accept() client websocket.state[http_client] response await client.get(https://www.example.com) await websocket.send_text(response.text) await websocket.close() app Starlette(lifespanlifespan, routes[WebSocketRoute(/ws, websocket_endpoint)])为什么属性式访问没有获得同等的类型推断原文档专门附了一段说明社区曾多次尝试让属性式访问request.state.http_client也获得同样的类型安全但始终没有令人满意的方案——要么会引入破坏性变更breaking changes要么受限于 Python 类型系统的能力边界typing limitations。因此最终选择了字典式访问作为类型安全的官方路径属性式访问仍然可用只是类型推断能力有限。仓库测试中也验证了这一能力在 tests/test_applications.py 中定义了CustomState类型的websocket_state与state_count端点分别通过websocket.state[count]与request.state[count]读取 lifespan 注入的状态并由 test_request_state 与 test_websocket_state 两个测试用例覆盖验证。在测试中运行 lifespan在单元测试中若直接使用TestClient(app)而不进入上下文lifespan 并不会被触发启动/关闭阶段的副作用也就不会执行。正确做法是把TestClient用作上下文管理器保证 lifespan 被调用from example import app from starlette.testclient import TestClient def test_homepage(): with TestClient(app) as client: # Applications lifespan is called on entering the block. response client.get(/) assert response.status_code 200 # And the lifespans teardown is run when exiting the block.时序说明注释即文档语义进入with块时触发应用的 lifespan 启动阶段进入yield之前此时共享状态已注入测试内发出的所有请求都能访问退出with块时运行 lifespan 的 teardownyield之后的清理代码此时资源释放完毕块外断言可验证清理副作用例如连接是否关闭、后台任务是否结束。这一机制在 starlette/testclient.py 中由TestClient.lifespan方法实现它构造{type: lifespan, state: ...}的 ASGI scope 调用应用并通过wait_startup/wait_shutdown两个协程分别等待lifespan.startup.complete或lifespan.startup.failed与lifespan.shutdown.complete或lifespan.shutdown.failed消息从而在上下文管理器的进入/退出边界完成与真实服务器一致的生命周期握手。仓库 tests/test_applications.py 中的test_app_async_cm_lifespan测试也验证了进入上下文前startup_complete为 False、退出上下文后清理完成的完整行为。参考路径速览官方文档docs/lifespan.md应用入口与lifespan参数传递starlette/applications.pylifespan 协议实现、_DefaultLifespan、旧式写法弃用starlette/routing.pyLifespan/StatelessLifespan/StatefulLifespan类型定义starlette/types.pyRequest.state属性实现starlette/requests.pyState类属性式与字典式双协议starlette/datastructures.pyTestClient生命周期握手实现starlette/testclient.py生命周期与状态共享的测试用例tests/test_applications.py【免费下载链接】starletteThe little ASGI framework that shines. 项目地址: https://gitcode.com/gh_mirrors/st/starlette创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考