Quick summary
Summarize this blog with AI
A reliable Requests call defines what success means before parsing the response: the expected HTTP status, media type, maximum body size, and data shape. It also supplies connection and read timeouts. This tutorial builds a small JSON client and exercises it against a local server, so the examples do not depend on a public API being available.
Tested with Python 3.12.14 and Requests 2.34.2. Run the blocks in order. The server serves synthetic responses on loopback only; no credentials or paid services are involved.
Read a bounded JSON response through a Session
A Session can reuse connections and defaults across calls. The function below makes one GET attempt, rejects redirects, validates the status and media type, and limits streamed bytes before decoding JSON. It does not turn network or parsing failures into an empty success result.
import json
import requests
from urllib.parse import urlsplit
def get_json(session, url, params=None, max_bytes=100_000, timeout=(5, 15)):
parts = urlsplit(url)
local_http = parts.scheme == 'http' and parts.hostname == '127.0.0.1'
if parts.scheme != 'https' and not local_http:
raise ValueError('use HTTPS except for a loopback test fixture')
if parts.username or parts.password or not parts.hostname:
raise ValueError('URL must have a host and no embedded credentials')
if type(max_bytes) is not int or max_bytes <= 0:
raise ValueError('max_bytes must be a positive integer')
with session.get(url, params=params, timeout=timeout, stream=True,
allow_redirects=False,
headers={'Accept': 'application/json', 'Accept-Encoding': 'identity'}) as response:
response.raise_for_status()
if response.status_code != 200:
raise ValueError('expected HTTP 200, not a redirect or empty response')
media_type = response.headers.get('Content-Type', '').split(';', 1)[0].lower()
if media_type != 'application/json' and not media_type.endswith('+json'):
raise ValueError('expected a JSON media type')
if response.headers.get('Content-Encoding', 'identity').lower() != 'identity':
raise ValueError('compressed responses are not accepted by this helper')
declared = response.headers.get('Content-Length')
if declared is not None and not 0 <= int(declared) <= max_bytes:
raise ValueError('declared response size exceeds the limit')
body = bytearray()
for chunk in response.iter_content(chunk_size=16384):
if len(body) + len(chunk) > max_bytes:
raise ValueError('response exceeds the byte limit')
body.extend(chunk)
return json.loads(bytes(body))
def item_names(payload):
if not isinstance(payload, dict) or not isinstance(payload.get('items'), list):
raise ValueError('expected an object containing an items list')
names = []
for item in payload['items']:
if not isinstance(item, dict) or not isinstance(item.get('name'), str):
raise ValueError('each item needs a text name')
names.append(item['name'])
return names
JSON decoding proves only that the body is parseable JSON. The separate item_names validator checks the part of the application schema this example consumes. It accepts an empty items list but rejects a missing field or non-text name. Real contracts may require identifiers, uniqueness, ranges, or additional fields.
Exercise success, malformed JSON and HTTP failure locally
from contextlib import contextmanager
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from threading import Thread
from urllib.parse import urlsplit
@contextmanager
def local_api():
hits = {'transient': 0}
class Handler(BaseHTTPRequestHandler):
def do_GET(self):
path = urlsplit(self.path).path
status, body = 200, b'{"items":[{"name":"Ada"},{"name":"Ben"}]}'
if path == '/missing':
status, body = 404, b'{"error":"missing"}'
elif path == '/bad-json':
body = b'not JSON'
elif path == '/transient':
hits['transient'] += 1
if hits['transient'] == 1:
status, body = 503, b'{"error":"try later"}'
self.send_response(status)
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
server = ThreadingHTTPServer(('127.0.0.1', 0), Handler)
thread = Thread(target=server.serve_forever, daemon=True)
thread.start()
try:
yield f'http://127.0.0.1:{server.server_port}', hits
finally:
server.shutdown()
server.server_close()
thread.join()
with local_api() as (base_url, hits), requests.Session() as session:
session.trust_env = False
payload = get_json(session, base_url + '/items', params={'limit': 2})
print(item_names(payload))
for path in ('/missing', '/bad-json'):
try:
get_json(session, base_url + path)
except (requests.HTTPError, json.JSONDecodeError) as exc:
print(type(exc).__name__)
Expected output:
['Ada', 'Ben']
HTTPError
JSONDecodeError
The fixture's limit parameter is only a demonstration of query encoding; this tiny server always returns two records. Real APIs define pagination and filtering rules themselves. Use params to encode query values, and use HTTPS for remote endpoints. The local server's plain HTTP is confined to 127.0.0.1.
Retry only a deliberately selected GET failure
This wrapper makes at most three total attempts and retries only HTTP 503 responses without Retry-After. It does not retry a malformed body, a 404, a timeout, or any POST. A Retry-After response is passed back to the caller so it can honor the service's delay rather than guessing. This intentionally small policy is easy to test and adapt to a documented API.
import time
def get_json_with_retry(session, url):
for attempt in range(3):
try:
return get_json(session, url)
except requests.HTTPError as exc:
response = exc.response
retryable = (
response is not None and response.status_code == 503
and 'Retry-After' not in response.headers
)
if not retryable or attempt == 2:
raise
time.sleep(0.1 * (2 ** attempt))
raise RuntimeError('unreachable retry state')
with local_api() as (base_url, hits), requests.Session() as session:
session.trust_env = False
print(item_names(get_json_with_retry(session, base_url + '/transient')))
print(hits['transient'])
Expected output:
['Ada', 'Ben']
2
Retrying even a GET assumes the endpoint behaves as a read operation. For state-changing requests, a timeout can mean the server completed the action but the client missed its response. Do not automatically resubmit a payment, message or job creation. Use the API's documented idempotency mechanism and reconciliation behavior when available.
Prepare JSON explicitly for a state-changing request
The following prepares a request without sending it. Passing json rather than data sets a JSON request body and media type. An actual send must add the endpoint's authentication, timeout, status and reconciliation contract.
prepared = requests.Request(
'POST', 'https://example.invalid/jobs', json={'job': 'daily-report'}
).prepare()
print(prepared.method)
print(prepared.headers['Content-Type'])
print(json.loads(prepared.body)['job'])
Expected output:
POST
application/json
daily-report
Make failure and resource ownership explicit
The response and Session context managers close their resources on success or failure. The fixture disables ambient proxy and netrc settings; a real deployment should configure any required proxy and credentials intentionally. Keep TLS verification enabled, avoid tokens in URLs, and do not log authorization headers.
A Requests timeout is a connection/read-inactivity rule, not a total elapsed deadline. Streaming read failures can also surface as ConnectionError or ChunkedEncodingError, so preserve the actual exception and job status. The tutorial's validation includes timeout, truncated body, oversized body, unexpected media type, persistent 503 and retry-limit cases.
For atomic file downloads, continue with download a file safely. Primary references: Requests response and error handling, Sessions and streaming, and HTTP idempotent methods.
