Module 7 of 9 · Lesson 19 of 26
Validate asynchronous provider responses
Work through validate asynchronous provider responses using a runnable reference, a focused regression check and a local extension.
Read in any order. All lessons stay open, including after an unanswered or incorrect check.
The integrated application owns one AsyncClient for its lifespan and closes it at shutdown. A request reuses that client for connection pooling instead of constructing a new one for every attempt. External waiting must not hold a capacity transaction open; the quote route does not start a reservation. The lessons use HTTPX MockTransport with synthetic responses, so no commercial provider, external account or actual email is needed. A local stub is an optional extension. Keep client-visible error messages stable and do not leak raw upstream bodies that may contain private values. Validation and transport recovery are separate decisions; retry policy comes next.
Run lesson19. One mocked response produces a typed decimal quote, while a negative amount is rejected as unusable. Inspect Quote in schemas.py and compare it with EquipmentIn. Read the application lifespan to see who owns and closes the client.
async def fetch_quote(client, url, *, attempts=3, deadline=0.5):
if not 1 <= attempts <= 5 or deadline <= 0:
raise ValueError("Choose 1–5 attempts and a positive deadline")
try:
async with asyncio.timeout(deadline):
for attempt in range(attempts):
try:
response = await client.get(url, timeout=min(deadline, 0.2))
if response.status_code in (502, 503, 504):
raise httpx.ReadError("Temporary upstream failure")
response.raise_for_status()
return Quote.model_validate(response.json())
except (httpx.ConnectError, httpx.ReadError, httpx.TimeoutException):
if attempt + 1 == attempts:
raise DomainError(
"provider_unavailable",
"The quote service is unavailable.",
503,
)
await asyncio.sleep(0.01 * 2**attempt)
except (httpx.HTTPStatusError, ValidationError, ValueError) as error:
raise DomainError(
"invalid_provider_response",
"The quote service returned an unusable response.",
502,
) from error
except TimeoutError as error:
raise DomainError(
"provider_deadline", "The quote service did not respond in time.", 504
) from errorpython run_checks.py -k lesson19Expected result The selected lesson test passes against a new temporary PostgreSQL database; the container is removed afterward.
Keep for reference
Equipment Rental lab and lesson checks
ZIP containing Python source, real Alembic migrations, 26 lesson checks, a dependency lock and text instructions. Extract it before following the local exercise.
Practise locally
Add tests for missing reference, unsupported currency and an unknown field. Confirm all fail at the provider boundary before any rental write. Build a local stub returning one valid quote if you want to inspect the HTTP exchange, but keep your automated tests on MockTransport so they are deterministic and offline.
The lesson check verifies the reference behavior. Add your own assertions for your change. Local practice is not uploaded or scored by this learning release.
Pause and reflect
What failure does this lesson prevent, and which assertion in lesson19 would expose it?
Use a concrete input, expected result and limitation from your local work. Saving a reflection does not certify the project.
Optional knowledge check
Is an HTTP 200 response sufficient to trust a provider quote?
No; validate the body against the expected schema as well.
Correct. Transport success and usable application data are different checks.
Yes; store any JSON immediately because the status succeeded.
Try another answer. A successful status can still carry malformed, unexpected or unsafe data.
Validate only the response status; ignore unexpected JSON fields.
Try another answer. Typed body validation prevents malformed provider data crossing the boundary.
This practice does not assess your project or award a certificate.
Your reading progress
Progress is saved in this browser when storage is available.