Complete worked example
python
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.testclient import TestClient
def inspect_example(requests):
events = []
@asynccontextmanager
async def lifespan(app):
app.state.marker = "ready"
events.append("open")
try:
yield
finally:
app.state.marker = None
events.append("close")
app = FastAPI(lifespan=lifespan)
@app.get("/ready")
def read():
return {"marker": getattr(app.state, "marker", None)}
with TestClient(app) as client:
observed = [client.get("/ready").json() for _ in range(requests)]
return {"observed": observed, "events": events,
"closed": getattr(app.state, "marker", None) is None}- Reasoning
- asynccontextmanager divides the lifespan into work before yield and cleanup after yield. The application stores its marker in state before serving the local requests. A finally block clears it and records closure even when the managed block exits through an exception. The TestClient with block triggers lifespan entry and exit. After leaving that block, the helper reports the response observations, event trace and final closed state. This is stronger evidence than checking a successful request alone: it also shows that the resource was acquired once and released.