Жизненный цикл

Приложение Starlette может зарегистрировать обработчик жизненного цикла для обработки кода, который должен быть запущен до запуска приложения или при его закрытии.

Импорт contextlib

from Starlette.applications import Starlette


Contextlib.asynccontextmanager
async def lifespan(app):
    async with some_async_resource ():
        «Run at Startup!»
        Yield
        «Run on shutdown!»


routes = [
    ...
]

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

Starlette не начинает обрабатывать входящие запросы до завершения жизненного цикла.

После того как все соединения отключены, а все текущие фоновые задачи выполнены, срок службы будет разобран.

Подумайте об использовании anyio.create_task_group() для управления асинхронными задачами.

Жизненный статус

Жизненный цикл имеет концепцию state, словарь, который можно использовать для совместного использования объектов между продолжительностью жизни и запросами.

Импорт contextlib
from typing import AsyncIterator, TypedDict

Import Tpx
from Starlette.applications import Starlette
Из starlette.requests import Request
from starlette.responses import PlainTextResponse
От Starlet.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)


Слайд (
    lifespan=lifespan,
    routes=[Route("/", Homepage)]
)

state, полученный в запросе, представляет собой мелкую копию состояния, полученного в обработчике жизненного цикла.

Срок работы во время тестирования

Вы должны использовать TestClient в качестве контекстного менеджера, чтобы гарантировать срок службы вызова.

Из example import app
Testclient import TestClient


def test_homepage ():
    with TestClient(приложение) 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.