DevOps NetworkDevOpsNetwork
Join Free
Practical infra coding

Aggregate paginated JSON from a local HTTP server

mediumAPIs and HTTP

Problem statement

A service catalog exposes GET /api/services?page=N. Each response is JSON with a services list (3 per page) and a next field holding the relative URL of the next page, or null on the last page. Write a script that walks every page with urllib.request and prints, per team, the total number of services and a count per status.

To keep the exercise offline and repeatable, the script starts its own server with http.server on 127.0.0.1, port 0 (the OS picks a free port), in a background thread. The server holds this data:

services (server side)

JSON
[
{"name": "auth", "team": "identity", "status": "healthy"},
{"name": "billing", "team": "payments", "status": "degraded"},
{"name": "checkout", "team": "payments", "status": "healthy"},
{"name": "ledger", "team": "payments", "status": "down"},
{"name": "login-ui", "team": "identity", "status": "healthy"},
{"name": "search", "team": "discovery"},
{"name": "recommend", "team": "discovery", "status": "healthy"}
]

Requirements:

  • Follow next until it is null, resolving it against the current URL with urljoin.
  • Set a timeout on every request and check that the response Content-Type is JSON before parsing it.
  • A service with no status counts as unknown; a service with no team goes under unowned.
  • Print teams in alphabetical order, statuses alphabetically within a team, then a needs attention: line listing teams that have anything down or unknown.
  • Finally, request /api/nope and show that the 404 is reported cleanly, not as a traceback.
  • Shut the server down at the end, even if something failed.

Examples

Example 1

Input: `python solution.py` (starts the local server, fetches 3 pages)

Output: discovery total=2 healthy=1, unknown=1 identity total=2 healthy=2 payments total=3 degraded=1, down=1, healthy=1 needs attention: discovery, payments bad path -> HTTP 404

Explanation: search has no status, so discovery shows unknown=1 and needs attention. payments has ledger down.

Hints

Approach

Optimal

Server. A BaseHTTPRequestHandler subclass slices the list by page number and returns JSON with an explicit Content-Type and Content-Length. Overriding log_message keeps request logs out of the output. The server runs serve_forever in a daemon thread, and the finally block calls shutdown() and server_close() so the port is released even if the client code raises.

Fetching. get_json sends an Accept header, always passes timeout, and uses with urlopen(...) so the connection is closed. It checks the Content-Type before calling json.load: a proxy error page or a login redirect often returns HTML with status 200, and json.load on HTML gives a confusing JSONDecodeError. fetch_services is a generator: it yields services page by page and follows next with urljoin(url, nxt), which handles both relative and absolute links. A page cap protects against a server that never says null.

Aggregating. summarize builds team -> Counter(status) in one pass, using .get() defaults for missing fields instead of letting one incomplete record crash the report. Printing sorts both levels, so the output does not depend on the order the API returned things in.

Errors. HTTPError is caught separately to report the status code. Anything else from the network (URLError, including connection refused and timeouts) or a non-JSON response ends up in the outer except with one readable line.

n is the number of services, t the number of teams and s the number of distinct statuses.

ComplexityTime O(n)Space O(t * s)
Python
import json
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlparse
SERVICES = [
{"name": "auth", "team": "identity", "status": "healthy"},
{"name": "billing", "team": "payments", "status": "degraded"},
{"name": "checkout", "team": "payments", "status": "healthy"},
{"name": "ledger", "team": "payments", "status": "down"},
{"name": "login-ui", "team": "identity", "status": "healthy"},
{"name": "search", "team": "discovery"},
{"name": "recommend", "team": "discovery", "status": "healthy"},
]
PER_PAGE = 3
class Handler(BaseHTTPRequestHandler):
# Local stand-in for the real service catalog: GET /api/services?page=N
def do_GET(self):
url = urlparse(self.path)
if url.path != "/api/services":
self.send_error(404)
return
page = int(parse_qs(url.query).get("page", ["1"])[0])
start = (page - 1) * PER_PAGE
chunk = SERVICES[start:start + PER_PAGE]
more = start + PER_PAGE < len(SERVICES)
body = json.dumps({
"services": chunk,
"next": f"/api/services?page={page + 1}" if more else None,
}).encode()
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def log_message(self, *args):
pass # keep the demo output clean
def start_server():
server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) # port 0 = any free port
threading.Thread(target=server.serve_forever, daemon=True).start()
return server
import json
from collections import Counter, defaultdict
from urllib.error import HTTPError, URLError
from urllib.parse import urljoin
from urllib.request import Request, urlopen
def get_json(url, timeout=5):
req = Request(url, headers={"Accept": "application/json"})
with urlopen(req, timeout=timeout) as resp:
ctype = resp.headers.get("Content-Type", "")
if not ctype.startswith("application/json"):
raise ValueError(f"{url}: expected JSON, got {ctype!r}")
return json.load(resp)
def fetch_services(base_url, first_path="/api/services", max_pages=100):
url = urljoin(base_url, first_path)
for _ in range(max_pages):
page = get_json(url)
yield from page.get("services", [])
nxt = page.get("next")
if not nxt:
return
url = urljoin(url, nxt) # works for relative and absolute links
raise RuntimeError("too many pages")
def summarize(services):
by_team = defaultdict(Counter)
for svc in services:
team = svc.get("team") or "unowned"
by_team[team][svc.get("status", "unknown")] += 1
return by_team
server = start_server()
base = f"http://127.0.0.1:{server.server_address[1]}"
try:
summary = summarize(fetch_services(base))
for team in sorted(summary):
counts = summary[team]
total = sum(counts.values())
detail = ", ".join(f"{s}={n}" for s, n in sorted(counts.items()))
print(f"{team:<10} total={total} {detail}")
bad = sorted(t for t, c in summary.items() if c["down"] or c["unknown"])
print("needs attention:", ", ".join(bad))
try:
get_json(base + "/api/nope")
except HTTPError as e:
print(f"bad path -> HTTP {e.code}")
except (URLError, ValueError) as e:
print("error:", e)
finally:
server.shutdown()
server.server_close()

Follow-up questions

  • Fetch the pages in parallel when the API returns a total page count up front.
  • The API starts requiring a bearer token. Where does it go, and how do you keep it out of logs?
  • Make the summary machine-readable with a --json flag so another tool can consume it.

Frequently asked questions

Status pages, service catalogs, CMDBs and monitoring tools all expose paginated JSON APIs, and small reporting scripts over them are everyday work. Interviewers use this to check that you handle pagination, timeouts, HTTP errors and messy records, not just the happy path.

It is not in the standard library. Many interview environments and minimal containers only have plain Python, and urllib.request covers GET requests, headers, timeouts and JSON with a few more lines. If requests is allowed, the structure of the solution stays the same.

A local server exercises the real HTTP path: headers, status codes, HTTPError and connection handling. It is still fully offline and deterministic, because it binds to 127.0.0.1 and serves fixed data.