Lifespan

Starlette applications can register a lifecycle handler to handle code that needs to run before the application starts or when the application closes.

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 does not begin processing any incoming requests until the lifecycle runs.

Once all connections have been closed and any ongoing background tasks have been completed, Lifetime Disassembly runs.

Consider using anyio.create_task_group() to manage asynchronous tasks.

life status

Lifetime has the concept of state, a dictionary that can be used to share objects between lifetimes and requests.

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

The state received on the request is a shallow copy of the state received on the lifetime handler.

Operational life under test

You should use TestClient as a context manager to ensure call life.

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.