HTTP and APIs
Fetch and parse data from an API endpoint
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.
To find it: print r.status_code before calling r.json(). A status
other than 200 means the response body is likely an error page, not
JSON. The JSONDecodeError is a symptom; the root cause is the failed
request that was never checked.
Submit a new record to an API as 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.
To find it: print r.request.headers["content-type"]. If it shows
application/x-www-form-urlencoded, the request was sent as form
data. Change data= to json= and print the header again to confirm
it now shows application/json.
Search an API with a multi-word query parameter
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.
To find it: print the constructed URL string before sending the
request. If a literal & appears inside a parameter value, the
ampersand was interpreted as a separator rather than being
percent-encoded. Pass parameters as a params= dictionary to let the
HTTP library encode them correctly.
Retrieve all records from a paginated API
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.
To find it: print len(records) after fetching and compare it to the
total field in the response JSON. If len(records) < total, the API
is returning data in pages. Check the response for a next_page,
next_cursor, or Link header that points to the next page.
Fetch data from a potentially slow API endpoint
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.
To find it: time the script with time python script.py. If it hangs
indefinitely, grep the source for timeout= and check whether the
value is None. Replace it with an explicit duration such as
timeout=10 and confirm the script now returns promptly with a
timeout error.
Create a new resource via a POST request
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.
To find it: print r.status_code after the POST. If it prints 201,
the resource was created successfully but the check r.status_code == 200
treats it as a failure. Replace the check with r.is_success to
accept any 2xx status code.
Retry a request after hitting a rate limit
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.
To find it: print r.status_code and r.headers.get("Retry-After")
on each attempt. If the status is 429 on every attempt and the
Retry-After value is nonzero, the script is not waiting before
retrying. Add time.sleep(int(r.headers["Retry-After"])) before the
next attempt.
Update a single field in an existing API resource
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.
To find it: fetch and print the resource before the request, then fetch and print it again after. If fields you did not include in the request body now show default or null values, PUT replaced the entire resource. Switch to PATCH and send only the fields you want to change.
Authenticate with an API using a secret key
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.
To find it: print str(r.request.url) to see the full URL including
query parameters. If the API key appears there (e.g., if
?api_key=secret is in the query) it will be recorded in any server
access log. Move it to an Authorization header instead.
Submit form data to an endpoint that redirects
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.
To find it: print r.request.method and r.request.url after the
call. If the final method is GET but you sent a POST, the redirect
changed the method. Print r.request.body to confirm the body was
dropped.