1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
|
Starlette applications can register a lifespan handler for dealing with
code that needs to run before the application starts up, or when the application
is shutting down.
```python
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 will not start serving any incoming requests until the lifespan has been run.
The lifespan teardown will run once all connections have been closed, and
any in-process background tasks have completed.
Consider using [`anyio.create_task_group()`](https://anyio.readthedocs.io/en/stable/tasks.html)
for managing asynchronous tasks.
## Lifespan State
The lifespan has the concept of `state`, which is a dictionary that
can be used to share the objects between the lifespan, and the requests.
```python
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 requests is a **shallow** copy of the state received on the
lifespan handler.
## Running lifespan in tests
You should use `TestClient` as a context manager, to ensure that the lifespan is called.
```python
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.
```
|
