正心全栈编程

管理应用生命周期 ​

有些资源只应在应用启动时准备，例如配置、初始数据或连接；关闭时还要释放它们。若把初始化放在每个处理函数里，请求一多就会重复执行。本专题用 Lifespan 管理这段共同的运行时间。

asynccontextmanager 中 yield 前的代码在启动阶段执行，yield 后的代码在关闭阶段执行。处理函数通过应用状态读取已准备好的数据。示例见下文。

启动与关闭时机 ​

先用最小日志验证启动、请求、关闭三个时机，避免把 Lifespan 误认为每次请求都会执行。

先写一个最小例子，只观察启动和关闭。

01_lifespan.py：

py
from contextlib import asynccontextmanager

from starlette.applications import Starlette
from starlette.responses import PlainTextResponse
from starlette.routing import Route


@asynccontextmanager
async def lifespan(app):
    print("应用启动：准备资源")
    yield
    print("应用关闭：清理资源")


async def homepage(request):
    return PlainTextResponse("首页")


routes = [
    Route("/", homepage),
]


app = Starlette(routes=routes, lifespan=lifespan)


if __name__ == "__main__":
    import uvicorn

    uvicorn.run("01_lifespan:app", host="127.0.0.1", port=8000, reload=True)

在 _code_starlette/p03c01 中运行 uvicorn 01_lifespan:app --reload。启动服务时，会先执行：

text
应用启动：准备资源

关闭服务时，会执行：

text
应用关闭：清理资源

而在应用运行期间，访问 / 不会反复执行这两段启动/关闭逻辑。

这里只需要记住：

  1. Lifespan 通过 lifespan=lifespan 挂到应用上。
  2. yield 前后对应启动和关闭两个阶段。
  3. Lifespan 不是每次请求都会执行。

Lifespan 完整生命周期时间线

共享应用状态 ​

初始化配置和文章数据后，通过 request.app.state 在处理函数中读取；只把应用范围内共享的状态放在这里。

现在把配置和文章数据放到启动阶段准备。

02_app_state.py：

py
from contextlib import asynccontextmanager

from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route


@asynccontextmanager
async def lifespan(app):
    app.state.config = {
        "app_name": "Starlette 文章服务",
        "version": "1.0.0",
    }

    app.state.articles = [
        {"id": 1, "title": "认识 Starlette 应用对象", "content": "示例内容"},
        {"id": 2, "title": "学习 Request 请求对象", "content": "示例内容"},
    ]

    print("应用启动：配置和文章数据已初始化")
    yield
    print("应用关闭：配置演示结束")


async def app_info(request):
    return JSONResponse(
        {
            "config": request.app.state.config,
            "article_count": len(request.app.state.articles),
        }
    )


routes = [
    Route("/app-info", app_info),
]


app = Starlette(routes=routes, lifespan=lifespan)


if __name__ == "__main__":
    import uvicorn

    uvicorn.run("02_app_state:app", host="127.0.0.1", port=8000, reload=True)

在示例目录改用 uvicorn 02_app_state:app --reload，访问 /app-info 后可看到初始化的数据。这里把两类数据放进了 app.state：

text
config
articles

然后在 endpoint 中通过：

python
request.app.state.config
request.app.state.articles

读取。

在启动时准备数据 ​

因为这些数据不是每个请求都要重新构造的。 它们更适合：

text
启动时准备一次
请求处理中直接使用

app.state 的使用边界 ​

app.state 适合放应用级共享资源。 但不要把所有临时变量都随手塞进去。

可以先记住：

text
适合放共享资源
不适合放一次性临时数据

app.state 应用级作用域与跨请求共享机制

清理初始化资源 ​

模拟资源关闭时使用 try/finally，确保退出阶段执行清理逻辑。

现在再加一个模拟资源，演示启动创建、关闭释放。

03_resource_cleanup.py：

py
from contextlib import asynccontextmanager

from starlette.applications import Starlette
from starlette.responses import JSONResponse, PlainTextResponse
from starlette.routing import Route


class FakeDatabase:
    def __init__(self):
        self.connected = False

    async def connect(self):
        self.connected = True
        print("FakeDatabase：连接已建立")

    async def close(self):
        self.connected = False
        print("FakeDatabase：连接已关闭")


@asynccontextmanager
async def lifespan(app):
    db = FakeDatabase()
    await db.connect()

    app.state.database = db
    app.state.articles = [
        {"id": 1, "title": "第一篇文章"},
        {"id": 2, "title": "第二篇文章"},
    ]

    try:
        print("应用启动：模拟资源已准备完成")
        yield
    finally:
        await db.close()
        print("应用关闭：模拟资源已释放")


async def app_info(request):
    db = request.app.state.database
    articles = request.app.state.articles
    return JSONResponse(
        {
            "db_connected": db.connected,
            "article_count": len(articles),
        }
    )


routes = [
    Route("/app-info", app_info),
]


app = Starlette(routes=routes, lifespan=lifespan)


if __name__ == "__main__":
    import uvicorn

    uvicorn.run("03_resource_cleanup:app", host="127.0.0.1", port=8000, reload=True)

这里的流程是：

text
启动时创建 FakeDatabase
↓
调用 connect()
↓
把连接对象存到 app.state.database
↓
请求中通过 request.app.state.database 使用它
↓
关闭时调用 close()

用 try/finally 保证清理 ​

因为关闭逻辑应尽量保证执行。 即使运行期间出现问题，也希望清理代码仍然能走到。

真实项目中的资源管理 ​

真实项目里，这种模式常见于：

text
数据库连接
缓存连接
消息队列连接
模型加载
配置加载

本章虽然只用了 FakeDatabase，但思路是一样的。

在示例目录运行 uvicorn 03_resource_cleanup:app --reload，访问 /app-info，再停止服务，观察连接与清理日志。python lifespan_demo.py 默认查看这一阶段的响应。

try...finally 资源初始化与优雅释放管道

Lifespan、中间件与 Endpoint ​

Lifespan 只在启动和关闭时执行 ​

它只在应用启动和关闭时执行。

如果某段逻辑需要：

text
启动时准备
关闭时清理

就优先考虑 Lifespan。

中间件处理每次请求 ​

如果某段逻辑需要：

text
每个请求都记录日志
每个请求都统计耗时
每个响应都加统一响应头

那它更适合中间件。

Endpoint 处理业务 ​

如果某段逻辑是：

text
读取文章列表
创建文章
修改文章
删除文章

那它就是 endpoint 的职责。

Lifespan、中间件与 Endpoint 职责分工矩阵

案例源码 ​

py
import requests


def test_minimal_lifespan_homepage():
    response = requests.get("http://127.0.0.1:8000/")
    print("\n=== 01_lifespan.py GET / ===")
    print(f"status_code: {response.status_code}")
    print(response.text)


def test_state_app_info():
    response = requests.get("http://127.0.0.1:8000/app-info")
    print("\n=== 02_app_state.py GET /app-info ===")
    print(f"status_code: {response.status_code}")
    print(response.json())


def test_fake_database_info():
    response = requests.get("http://127.0.0.1:8000/app-info")
    print("\n=== 03_resource_cleanup.py GET /app-info ===")
    print(f"status_code: {response.status_code}")
    print(response.json())


if __name__ == "__main__":
    # 启动 03_resource_cleanup.py 后运行这一项；其他函数可在相应阶段单独调用。
    test_fake_database_info()