HTTP and APIs
No Status Check Before Parsing
Run the script and read the error message. What type of error is raised, and which line causes it? Is the problem in the line that fails or somewhere earlier?
import asyncio
import httpx
async def app(scope, receive, send):
assert scope["type"] == "http"
await send({
"type": "http.response.start",
"status": 404,
"headers": [(b"content-type", b"text/html")],
})
await send({
"type": "http.response.body",
"body": b"<html><body><h1>404 Not Found</h1></body></html>",
})
async def main():
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
r = await client.get("/data")
data = r.json()
print(data)
asyncio.run(main())
Show explanation
The bug is calling .json() without first checking r.status_code. A 404 response
returns an HTML error page, not JSON, so .json() raises a JSONDecodeError at the
parsing step rather than flagging the real problem, which is the failed request.
Shows how to check r.status_code or call r.raise_for_status() before reading
the response body.
Form Encoding Instead of JSON
Run the script and read the output. What content type does the server report receiving? What content type did you intend to send?
import asyncio
import httpx
async def app(scope, receive, send):
assert scope["type"] == "http"
event = await receive()
headers = dict(scope["headers"])
content_type = headers.get(b"content-type", b"(not set)").decode()
body = event.get("body", b"")
msg = f"content-type: {content_type}\nbody: {body!r}".encode()
await send({
"type": "http.response.start",
"status": 200,
"headers": [(b"content-type", b"text/plain")],
})
await send({"type": "http.response.body", "body": msg})
async def main():
payload = {"name": "Alice", "score": 95}
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
r = await client.post("/submit", data=payload)
print(r.text)
asyncio.run(main())
Show explanation
The bug is using data= instead of json= in the POST call. data= sends a
form-encoded body with content-type: application/x-www-form-urlencoded, while
most APIs expect json= which sends a JSON body with content-type:
application/json. Shows the difference between these two keyword arguments and
why the server may silently reject or misparse a request sent with the wrong
encoding.
Special Characters in Query Parameters
Run the script and read the output. How many query parameters did the server receive? How many did the code intend to send?
import asyncio
import httpx
async def app(scope, receive, send):
assert scope["type"] == "http"
qs = scope.get("query_string", b"").decode()
params = {}
for pair in qs.split("&"):
if "=" in pair:
k, v = pair.split("=", 1)
params[k] = v
elif pair:
params[pair] = "(no value)"
msg = f"parsed params: {params}".encode()
await send({
"type": "http.response.start",
"status": 200,
"headers": [(b"content-type", b"text/plain")],
})
await send({"type": "http.response.body", "body": msg})
async def main():
category = "books&games"
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
r = await client.get(f"/items?category={category}&limit=10")
print(r.text)
asyncio.run(main())
Show explanation
The bug is embedding a value that contains & directly in an f-string URL. The
ampersand is interpreted as a query-string separator, so the server receives
category=books and a bare key games instead of category=books&games. Shows
how to pass query parameters as a params= dict so the HTTP client encodes special
characters correctly.
Missing Pagination
Run the script and compare the number of records retrieved to the total available. What field in the response tells you that more data exists?
import asyncio
import json
import httpx
ALL_RECORDS = [{"id": i, "value": i * 10} for i in range(1, 21)]
PAGE_SIZE = 5
async def app(scope, receive, send):
assert scope["type"] == "http"
qs = scope.get("query_string", b"").decode()
params = dict(p.split("=", 1) for p in qs.split("&") if "=" in p)
page = int(params.get("page", "1"))
start = (page - 1) * PAGE_SIZE
items = ALL_RECORDS[start : start + PAGE_SIZE]
next_page = page + 1 if start + PAGE_SIZE < len(ALL_RECORDS) else None
body = json.dumps({"items": items, "next_page": next_page}).encode()
await send({
"type": "http.response.start",
"status": 200,
"headers": [(b"content-type", b"application/json")],
})
await send({"type": "http.response.body", "body": body})
async def main():
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
r = await client.get("/records?page=1")
data = r.json()
print(f"retrieved {len(data['items'])} records")
print(f"next_page field in response: {data['next_page']!r}")
print(f"total records on server: {len(ALL_RECORDS)}")
asyncio.run(main())
Show explanation
The bug is fetching only the first page and ignoring the next_page field in the
response. No error is raised; the script silently processes a fraction of the
available data. Shows how to recognize and follow pagination cursors and why APIs
return data in pages rather than all at once.
Disabled Request Timeout
Run the script. Does it return promptly? How long does it wait before producing output?
import asyncio
import httpx
# Simulated server delay in seconds
SERVER_DELAY = 10
async def app(scope, receive, send):
assert scope["type"] == "http"
await asyncio.sleep(SERVER_DELAY)
await send({
"type": "http.response.start",
"status": 200,
"headers": [(b"content-type", b"application/json")],
})
await send({"type": "http.response.body", "body": b'{"status": "done"}'})
async def main():
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
r = await client.get("/report", timeout=None)
print(r.json())
asyncio.run(main())
Show explanation
The bug is passing timeout=None, which disables all timeouts and causes the call to
wait indefinitely for a slow or unresponsive server. Shows the difference between
httpx's default timeout and timeout=None, and how to set an explicit
httpx.Timeout to bound how long a request may take.
Checking for 200 Instead of Any 2xx
Run the script. What status code does the server return? What does the script print? Is the request actually successful?
import asyncio
import json
import httpx
async def app(scope, receive, send):
assert scope["type"] == "http"
event = await receive()
body = json.loads(event.get("body", b"{}"))
resource = {"id": 42, **body}
await send({
"type": "http.response.start",
"status": 201,
"headers": [(b"content-type", b"application/json")],
})
await send({
"type": "http.response.body",
"body": json.dumps(resource).encode(),
})
async def main():
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
r = await client.post("/users", json={"name": "Alice"})
if r.status_code == 200:
print("created:", r.json())
else:
print(f"request failed with status {r.status_code}")
asyncio.run(main())
Show explanation
The bug is comparing r.status_code == 200 when a successful POST returns
201 Created. The response is treated as a failure even though the resource was
created. Shows the range of 2xx status codes, when each is used, and how to use
r.is_success to accept any successful response.
Retrying Without Reading Retry-After
Run the script and look at the status code and headers on each attempt. What does
the Retry-After header contain? Does the script wait before retrying?
import asyncio
import json
import httpx
_request_count = 0
async def app(scope, receive, send):
global _request_count
assert scope["type"] == "http"
_request_count += 1
await send({
"type": "http.response.start",
"status": 429,
"headers": [
(b"content-type", b"application/json"),
(b"retry-after", b"60"),
],
})
await send({
"type": "http.response.body",
"body": json.dumps({"error": "rate limit exceeded"}).encode(),
})
async def main():
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
for attempt in range(1, 4):
r = await client.get("/api/data")
if r.status_code == 200:
print("success:", r.json())
break
print(f"attempt {attempt}: status {r.status_code}, retry-after={r.headers.get('retry-after')!r}")
print(f"total requests sent: {_request_count}")
asyncio.run(main())
Show explanation
The bug is retrying immediately after a 429 Too Many Requests response without
reading the Retry-After header. Each retry is rejected for the same reason and the
script loops without ever succeeding. Shows what 429 means, how to detect it, and
how to wait the server-specified duration before the next attempt.
PUT Replaces Instead of Updates
Run the script and compare the resource state before and after the PUT request. Which fields changed, and which fields were you expecting to keep?
import asyncio
import json
import httpx
# Server-side resource state
_resource = {"name": "Alice", "email": "alice@example.com", "role": "admin"}
async def app(scope, receive, send):
assert scope["type"] == "http"
event = await receive()
if scope["method"] == "PUT":
update = json.loads(event.get("body", b"{}"))
# PUT replaces the entire resource with the request body
_resource.clear()
_resource.update(update)
body = json.dumps(_resource, indent=2).encode()
await send({
"type": "http.response.start",
"status": 200,
"headers": [(b"content-type", b"application/json")],
})
await send({"type": "http.response.body", "body": body})
async def main():
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
r = await client.get("/user/1")
print("before:", r.text)
r = await client.put("/user/1", json={"name": "Alicia"})
print("after:", r.text)
asyncio.run(main())
Show explanation
The bug is using PUT with a partial body. PUT replaces the entire resource with the request body, so any field not included in the request is wiped. Shows the semantic difference between PUT (full replacement) and PATCH (partial update), and why sending only the fields you want to change requires PATCH.
API Key in URL Query String
Run the script and read the output. In which part of the request does the API key appear? What would a server log entry look like?
import asyncio
import httpx
async def app(scope, receive, send):
assert scope["type"] == "http"
qs = scope.get("query_string", b"").decode()
headers = dict(scope["headers"])
auth = headers.get(b"authorization", b"(not set)").decode()
msg = f"query string : {qs!r}\nauthorization: {auth!r}".encode()
await send({
"type": "http.response.start",
"status": 200,
"headers": [(b"content-type", b"text/plain")],
})
await send({"type": "http.response.body", "body": msg})
async def main():
API_KEY = "secret-key-12345"
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
r = await client.get(f"/data?api_key={API_KEY}")
print(r.text)
asyncio.run(main())
Show explanation
The bug is placing the API key in the URL query string. Query strings are recorded
in server access logs, browser history, and any intermediate proxies, so the secret
is exposed in plaintext. Shows how to pass credentials in an Authorization
header instead, where they are kept out of logs and not cached by browsers.
POST Body Lost After Redirect
Run the script and read the output. What HTTP method reached the final endpoint?
What happened to the request body that was sent to /submit?
import asyncio
import json
import httpx
async def app(scope, receive, send):
assert scope["type"] == "http"
if scope["path"] == "/submit":
await send({
"type": "http.response.start",
"status": 302,
"headers": [(b"location", b"http://test/result")],
})
await send({"type": "http.response.body", "body": b""})
else:
event = await receive()
body = event.get("body", b"")
msg = json.dumps({
"method": scope["method"],
"body": body.decode() if body else "(empty)",
}, indent=2).encode()
await send({
"type": "http.response.start",
"status": 200,
"headers": [(b"content-type", b"application/json")],
})
await send({"type": "http.response.body", "body": msg})
async def main():
payload = {"username": "alice", "password": "s3cr3t"}
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
r = await client.post("/submit", json=payload, follow_redirects=True)
print(r.text)
asyncio.run(main())
Show explanation
The bug is following a 302 redirect from a POST request. HTTP convention changes the method from POST to GET when following a 302, so the request body is silently dropped and the final endpoint receives an empty GET request. Shows how redirects interact with request methods, and how to detect this by inspecting the method and body at the redirected URL.