生命周期

Starlette 应用程序可以注册一个生命周期处理程序，用于处理在应用程序启动前或应用程序关闭时需要运行的代码。

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)

在生命周期运行之前，Starlette 不会开始处理任何传入请求。

一旦所有连接都已关闭，且任何正在进行的后台任务都已完成，将运行使用寿命拆解。

考虑使用 anyio.create_task_group() 来管理异步任务。

寿命状态

寿命具有 state 的概念，这是一个字典，可用于在寿命和请求之间共享对象。

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)]
)

在请求上接收到的 state 是在寿命处理程序上接收到的状态的浅拷贝。

测试中的运行寿命

您应将 TestClient 用作上下文管理器，以确保调用寿命。

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.