企业官网建设流程全解析

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(routes=routes, lifespan=lifespan)

要点解读：

• yield之前的代码块在应用启动阶段执行（对应打印 "Run at startup!"）；yield之后的代码块在应用关闭阶段执行（对应打印 "Run on shutdown!"）。
• 通过Starlette(..., lifespan=lifespan)把该处理器挂载到应用上，函数签名中的app参数即当前应用实例，可用于读取app.state等。
• 这种结构天然适配"随用随建、随关随毁"的资源管理模式，例如async with some_async_resource():可以换成任何支持异步上下文协议的资源。

生命周期时序的两条硬性保证

原文档明确了两条重要的时序语义：

1. 请求不会被提前处理：Starlette 在 lifespan 运行完成之前，不会开始处理任何进入的请求。也就是说，yield真正"放行"了请求服务；如果启动阶段抛错或未完成，应用不会对外提供服务。
2. 关闭在一切收尾之后：lifespan 的 teardown（yield之后的代码）会在所有连接都已关闭、所有进程内后台任务都已完成之后才执行。这保证了你关闭连接池时不会再收到新的请求或后台任务占用。

对于需要在后台维护异步任务的场景，原文档建议优先使用anyio.create_task_group()来管理这些异步任务，使其纳入同一生命周期管理范围，避免任务游离于应用之外。

源码层面的实现印证

lifespan参数最终从Starlette.__init__传入内部的Router（见 starlette/applications.py 的构造函数与self.router = Router(routes, lifespan=lifespan)）。真正的执行逻辑在 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。
• 若应用没有显式注册 lifespan，Router.__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( lifespan=lifespan, 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 的完整流转路径：

1. lifespan 中yield出的字典，在 starlette/routing.py 中被合并进 ASGI scope 的state字段：scope["state"].update(maybe_state)。
2. 当请求到达时，Request.state属性（见 starlette/requests.py）会基于同一个scope["state"]字典构造State对象：self._state = State(self.scope["state"])，从而让请求侧读到与 lifespan 相同的内容。
3. State类的实现位于 starlette/datastructures.py，它把内部数据存放在self._state字典中，并同时实现了属性式访问（__getattr__/__setattr__）与字典式访问（__getitem__/__setitem__）两套协议，这也正是下一节两种访问语法都能工作的底层原因。

访问 State：属性式与字典式两种语法

state既可以用属性式语法访问（request.state.foo），也可以用字典式语法访问（request.state["foo"]）。

字典式语法是在Starlette 0.52.0（2026 年 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(lifespan=lifespan, 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(lifespan=lifespan, 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: # Application's lifespan is called on entering the block. response = client.get("/") assert response.status_code == 200 # And the lifespan's teardown is run when exiting the block.

时序说明（注释即文档语义）：

• 进入with块时：触发应用的 lifespan 启动阶段（进入yield之前），此时共享状态已注入，测试内发出的所有请求都能访问；
• 退出with块时：运行 lifespan 的 teardown（yield之后的清理代码），此时资源释放完毕，块外断言可验证清理副作用（例如连接是否关闭、后台任务是否结束）。

这一机制在 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.py
• lifespan 协议实现、_DefaultLifespan、旧式写法弃用：starlette/routing.py
• Lifespan/StatelessLifespan/StatefulLifespan类型定义：starlette/types.py
• Request.state属性实现：starlette/requests.py
• State类（属性式与字典式双协议）：starlette/datastructures.py
• TestClient生命周期握手实现：starlette/testclient.py
• 生命周期与状态共享的测试用例：tests/test_applications.py

【免费下载链接】starletteThe little ASGI framework that shines. 🌟项目地址: https://gitcode.com/gh_mirrors/st/starlette