Cycle de vie

Les applications Starlette peuvent enregistrer un gestionnaire de cycle de vie qui traite le code qui doit être exécuté avant le démarrage de l'application ou lorsque l'application est fermée.

Importer le contexte lib

de starlette.applications import Starlette


@contextlib.asynccontextmanager
async def lifespan(app):
    async avec some_async_resource():
        print("Run at start-up!")
        Yield est un
        print("Run on shutdown!")


Routes = [
    ...
]

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

Starlette ne commence pas à traiter les demandes entrantes tant que le cycle de vie n'est pas exécuté.

Une fois que toutes les connexions ont été fermées et que toutes les tâches en arrière-plan en cours ont été terminées, le démontage de la durée de vie est exécuté.

Utilisez anyio.create_task_group() pour gérer les tâches asynchrones.

État de vie

La durée de vie a le concept de state, un dictionnaire qui peut être utilisé pour partager des objets entre la durée de vie et la demande.

Importer le contexte lib
from typing import AsyncIterator, TypedDict

Import httpx
de starlette.applications import Starlette
de starlette.requests Demande d'importation
from starlette.responses import PlainTextResponse
de starlette.routing Route d'importation


État de la classe (TypedDict):
    http_client: httpx.AsyncClient


@contextlib.asynccontextmanager
async def lifespan(app: Starlette) -> AsyncIterator[State]:
    async avec httpx.AsyncClient() comme client:
        Yield, le {"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)]
)

Le state reçu sur la requête est une copie superficielle de l'état reçu sur le gestionnaire de durée de vie.

La vie opérationnelle en test

Vous devriez utiliser TestClient comme gestionnaire de contexte pour assurer la longévité des appels.

De l'exemple de l'application d'importation
depuis starlette.testclient import TestClient


Def test_homepage():
    Avec TestClient(app) comme client:
        Le lifespan de l'application est appelé à enterrer le bloc.
        response = client.get("/")
        assert response.status_code == 200

    #Et l'apprentissage du lifespan est en cours d'exécution quand il existe le bloc.