Bounded HTTP reads and JSON documents

pythonapi-clientsautomationindependent-practice

Saved on this device

Full lesson · 4 learning steps Open any exercise for focused practice

Step 1 of 4 · intro card

Bounded HTTP reads and JSON documents

Concept and contract

Read a controlled HTTP response with an explicit timeout, byte budget, media type and parse boundary.

A successful connection is only the beginning of reading an API response. The client also needs a timeout, a bounded read, a declared media type and a parsing policy.

This worked example accepts a JSON object from a caller-selected authorized endpoint. The downloadable lab supplies temporary loopback URLs only. The function is not a network sandbox and must not be handed arbitrary untrusted destinations in a real service.

Mechanism and worked example

The opener disables ambient proxies and automatic redirects for this small controlled client.

The opener disables ambient proxies and automatic redirects for this small controlled client. It uses a context manager to close the response, checks the media type and reads at most the budget plus one byte.

That extra byte detects overflow without reading the whole body. UTF-8 decoding, strict rejection of non-finite JSON constants and a top-level object check then establish successive representation boundaries before the result is returned.

Complete worked example

python

from urllib.request import build_opener, HTTPRedirectHandler, ProxyHandler

class NoRedirect(HTTPRedirectHandler):
    def redirect_request(self, request, file, code, message, headers, new_url):
        return None

import json

def invalid_constant(value):
    raise ValueError("non-finite JSON constant")

def json_document(url, max_bytes=1024):
    if type(max_bytes) is not int or not 1 <= max_bytes <= 4096:
        raise ValueError("byte budget required")
    opener = build_opener(ProxyHandler({}), NoRedirect())
    with opener.open(url, timeout=2) as response:
        if response.headers.get_content_type() != "application/json":
            raise ValueError("JSON media type required")
        raw = response.read(max_bytes + 1)
    if len(raw) > max_bytes:
        raise ValueError("response exceeds budget")
    result = json.loads(raw.decode("utf-8"), parse_constant=invalid_constant)
    if not isinstance(result, dict):
        raise ValueError("object required")
    return result

Reasoning: Read a controlled HTTP response with an explicit timeout, byte budget, media type and parse boundary. Compare the stated values and failure boundary before changing the implementation.

Counterexamples and common mistakes

The counterexample reads the whole response before checking its length.

The counterexample reads the whole response before checking its length. It can return the same error as the bounded version while already consuming excessive memory.

Content-Length alone is not a substitute for enforcing the read bound. The lab therefore records the read size separately from its value assertions. A timeout here bounds blocking operations rather than proving a whole-workflow deadline, and rejected HTTP responses still need their own error policy.

Counterexample: a runnable policy defect

python

from urllib.request import build_opener, HTTPRedirectHandler, ProxyHandler

class NoRedirect(HTTPRedirectHandler):
    def redirect_request(self, request, file, code, message, headers, new_url):
        return None

import json

def invalid_constant(value):
    raise ValueError("non-finite JSON constant")

def json_document(url, max_bytes=1024):
    if type(max_bytes) is not int or not 1 <= max_bytes <= 4096:
        raise ValueError("byte budget required")
    opener = build_opener(ProxyHandler({}), NoRedirect())
    with opener.open(url, timeout=2) as response:
        if response.headers.get_content_type() != "application/json":
            raise ValueError("JSON media type required")
        raw = response.read()
    if len(raw) > max_bytes:
        raise ValueError("response exceeds budget")
    result = json.loads(raw.decode("utf-8"), parse_constant=invalid_constant)
    if not isinstance(result, dict):
        raise ValueError("object required")
    return result

Reasoning: This runnable counterexample violates the declared contract; the companion lab requires it to fail.

Independent practice: API client task

Require an items field whose value is a list, after the document has passed its existing checks.

Require an items field whose value is a list, after the document has passed its existing checks. Start with an empty list, then reject a missing field and a string pretending to be the list.

Keep the byte bound, timeout and response closure intact. This adds one application-shape rule on top of a JSON document boundary without requiring a full schema package.

Step 1: Require an items list in the parsed JSON object. Step 2: Preserve the bounded read and closure behavior.

Practice starter: solve before reading the solution

python

from urllib.request import build_opener, HTTPRedirectHandler, ProxyHandler

class NoRedirect(HTTPRedirectHandler):
    def redirect_request(self, request, file, code, message, headers, new_url):
        return None

import json

def invalid_constant(value):
    raise ValueError("non-finite JSON constant")

def json_document(url, max_bytes=1024):
    # Remove this placeholder, then adapt the retained baseline:
    # Require an items list in the parsed JSON object.
    raise NotImplementedError
    if type(max_bytes) is not int or not 1 <= max_bytes <= 4096:
        raise ValueError("byte budget required")
    opener = build_opener(ProxyHandler({}), NoRedirect())
    with opener.open(url, timeout=2) as response:
        if response.headers.get_content_type() != "application/json":
            raise ValueError("JSON media type required")
        raw = response.read(max_bytes + 1)
    if len(raw) > max_bytes:
        raise ValueError("response exceeds budget")
    result = json.loads(raw.decode("utf-8"), parse_constant=invalid_constant)
    if not isinstance(result, dict):
        raise ValueError("object required")
    return result

Reasoning: Require an items list in the parsed JSON object. Preserve the bounded read and closure behavior.

Hint 1: Validate the decoded result after checking that it is a dict.
Hint 2: An empty list is valid; a missing key or string value is not.

Solution, tests and trade-offs

The reference suite exercises valid JSON, an oversized body, an incorrect media type, an invalid top-level array, a non-finite constant and a redirect.

The reference suite exercises valid JSON, an oversized body, an incorrect media type, an invalid top-level array, a non-finite constant and a redirect. A separate instrumented response verifies the requested read size and confirms closure, so an unbounded read cannot hide behind an eventual error.

Practice adds items-shape failures. The fixtures never contact the documentation hostname, real company feeds or the ApplyDjinn production API.

Practice solution and acceptance checks

python

from urllib.request import build_opener, HTTPRedirectHandler, ProxyHandler

class NoRedirect(HTTPRedirectHandler):
    def redirect_request(self, request, file, code, message, headers, new_url):
        return None

import json

def invalid_constant(value):
    raise ValueError("non-finite JSON constant")

def json_document(url, max_bytes=1024):
    if type(max_bytes) is not int or not 1 <= max_bytes <= 4096:
        raise ValueError("byte budget required")
    opener = build_opener(ProxyHandler({}), NoRedirect())
    with opener.open(url, timeout=2) as response:
        if response.headers.get_content_type() != "application/json":
            raise ValueError("JSON media type required")
        raw = response.read(max_bytes + 1)
    if len(raw) > max_bytes:
        raise ValueError("response exceeds budget")
    result = json.loads(raw.decode("utf-8"), parse_constant=invalid_constant)
    if not isinstance(result, dict):
        raise ValueError("object required")
    if not isinstance(result.get("items"), list):
        raise ValueError("items list required")
    return result

Reasoning: Validate the decoded result after checking that it is a dict. An empty list is valid; a missing key or string value is not.

Interview reasoning and next step

Explain why representation validation has layers: status and headers, bounded bytes, decoded text, parsed JSON and domain shape.

Explain why representation validation has layers: status and headers, bounded bytes, decoded text, parsed JSON and domain shape. A compressed response needs a deliberate policy for decompression and post-decompression limits; this example does not request or implement compressed payloads.

A production client also needs an authorized origin policy and lifecycle budget. Keep those responsibilities explicit when composing this helper into a larger importer.

Optional stretch — after the core task works: Design separate compressed-wire and decompressed-body byte budgets.

Computer practice: https://applydjinn.com/typorithm/python-automation-lab-20261105.zip. Run verify.py using your selected Python interpreter, then adapt the retained exercise baseline and check it with --practice followed by the lesson id. The standard-library lab uses owned temporary 127.0.0.1 loopback servers, fictional pages and disposable files; it never scans production feeds or opens the working ApplyDjinn database.

Read a controlled HTTP response with an explicit timeout, byte budget, media type and parse boundary.