Networking and HTTP Clients

What we're doing: Make the same GET request with all three libraries and confirm each one actually reaches httpbin.org and gets the same query parameters back.

import asyncio
import requests
import httpx
import aiohttp

r = requests.get('https://httpbin.org/get', params={'q': '1'})
print('requests:', r.status_code, r.json()['args'])

r = httpx.get('https://httpbin.org/get', params={'q': '1'})
print('httpx:', r.status_code, r.json()['args'])

async def aiohttp_get():
    async with aiohttp.ClientSession() as session:
        async with session.get('https://httpbin.org/get', params={'q': '1'}) as resp:
            data = await resp.json()
            return resp.status, data['args']

print('aiohttp:', asyncio.run(aiohttp_get()))

6

requests.get is a module-level function — no client object to construct for a single call.

9

httpx.get mirrors requests' call shape exactly — params=, .status_code, .json() all match.

13

aiohttp requires both an outer ClientSession and an inner response context manager, and every read (.json()) is awaited.

requests: 200 {'q': '1'}
httpx: 200 {'q': '1'}
aiohttp: (200, {'q': '1'})

Why this works: All three libraries reach the same server and get the same echoed query parameters back, confirming the API differences are about call shape and concurrency model — not about what HTTP they can perform. requests and httpx share almost identical sync call shapes; aiohttp's two-level async context manager is the one genuinely different pattern to learn.

What we're doing: Trigger a real timeout against a genuinely slow endpoint, then trigger a real exhausted-retries error against a genuinely failing endpoint, proving both failure modes actually fire rather than assuming the configuration works.

import time
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# 1. Timeout: httpbin.org/delay/3 sleeps 3s server-side before responding
try:
    requests.get('https://httpbin.org/delay/3', timeout=1)
except requests.exceptions.Timeout as e:
    print('Timeout raised:', type(e).__name__)

# 2. Retries: httpbin.org/status/500 always returns 500 -- retries exhaust
retry_strategy = Retry(
    total=3,
    backoff_factor=0.1,
    status_forcelist=[500, 502, 503, 504],
    allowed_methods=['GET'],
)
session = requests.Session()
session.mount('https://', HTTPAdapter(max_retries=retry_strategy))

start = time.time()
try:
    session.get('https://httpbin.org/status/500')
except requests.exceptions.RetryError as e:
    elapsed = time.time() - start
    print('RetryError after exhausting retries:', type(e).__name__, 'elapsed:', round(elapsed, 2))

8

timeout=1 means "give up after 1 second" -- the server deliberately takes 3, so this always times out.

20

total=3 with backoff_factor=0.1 means up to 3 retries with increasing delay between them, only for the listed status codes.

24

Every one of the 4 attempts (1 original + 3 retries) gets a 500, so urllib3 raises RetryError once total is exhausted.

Timeout raised: ReadTimeout
RetryError after exhausting retries: RetryError elapsed: 2.89

Why this works: Both failures are real, not simulated: httpbin.org/delay/3 genuinely sleeps 3 seconds server-side, so a 1-second timeout genuinely expires client-side. httpbin.org/status/500 genuinely returns 500 every time, so all 4 attempts (backed off with increasing delay, hence the 2.89s elapsed) genuinely fail, and urllib3 genuinely raises RetryError once the retry budget hits zero — this is what "exhausted" looks like in practice, not a theoretical ceiling.

What we're doing: Send a real Bearer token and real Basic auth credentials against httpbin.org's dedicated auth-check endpoints and confirm both are accepted.

import requests
import httpx

r = requests.get('https://httpbin.org/bearer', headers={'Authorization': 'Bearer test-token-123'})
print('bearer:', r.status_code, r.json())

r = requests.get('https://httpbin.org/basic-auth/user/pass', auth=('user', 'pass'))
print('requests basic:', r.status_code, r.json())

r = httpx.get('https://httpbin.org/basic-auth/user/pass', auth=httpx.BasicAuth('user', 'pass'))
print('httpx basic:', r.status_code, r.json())

r = requests.get('https://httpbin.org/headers', headers={'X-Client-Id': 'svc-42', 'Authorization': 'Bearer abc'})
echoed = r.json()['headers']
print('echoed:', echoed.get('X-Client-Id'), echoed.get('Authorization'))

4

/bearer validates the Authorization header against the Bearer scheme and echoes the token back if accepted.

7

requests' auth=('user', 'pass') tuple builds the Basic auth header (base64 of user:pass) automatically.

10

httpx has no auth= tuple shortcut -- httpx.BasicAuth(...) is the explicit equivalent.

bearer: 200 {'authenticated': True, 'token': 'test-token-123'}
requests basic: 200 {'authenticated': True, 'user': 'user'}
httpx basic: 200 {'authenticated': True, 'user': 'user'}
echoed: svc-42 Bearer abc

Why this works: httpbin.org's /bearer and /basic-auth endpoints actually validate the credentials sent, not just accept anything — a 200 with 'authenticated': True confirms the header was built and sent correctly, not merely that a request was made. The last call confirms custom and auth headers travel together, unmodified, exactly as sent.

What we're doing: Stream a real 2048-byte response from all three libraries in fixed 512-byte chunks and confirm the total bytes and chunk count match what a whole-body read would give, proving nothing is silently dropped or duplicated.

import asyncio
import requests
import httpx
import aiohttp

with requests.get('https://httpbin.org/stream-bytes/2048', stream=True) as r:
    total = chunks = 0
    for chunk in r.iter_content(chunk_size=512):
        total += len(chunk)
        chunks += 1
    print('requests:', total, 'bytes in', chunks, 'chunks')

with httpx.stream('GET', 'https://httpbin.org/stream-bytes/2048') as r:
    total = chunks = 0
    for chunk in r.iter_bytes(512):
        total += len(chunk)
        chunks += 1
    print('httpx:', total, 'bytes in', chunks, 'chunks')

async def aiohttp_stream():
    total = chunks = 0
    async with aiohttp.ClientSession() as session:
        async with session.get('https://httpbin.org/stream-bytes/2048') as resp:
            async for chunk in resp.content.iter_chunked(512):
                total += len(chunk)
                chunks += 1
    return total, chunks

print('aiohttp:', asyncio.run(aiohttp_stream()))

5

stream=True is required -- without it, requests downloads the entire body before get() even returns.

13

httpx.stream() is itself the context manager; the response inside it is only partially read until you iterate.

22

aiohttp streams by default -- .content is already a StreamReader; iter_chunked just controls the chunk size.

requests: 2048 bytes in 4 chunks
httpx: 2048 bytes in 4 chunks
aiohttp: (2048, 4)

Why this works: All three report the same 2048 total bytes across 4 chunks of 512 -- the endpoint's real byte count, confirming chunking splits the SAME data rather than losing or duplicating any of it. requests and httpx needed an explicit opt-in (stream=True, or the .stream() context manager); aiohttp streams by default because .content is always a StreamReader.

What we're doing: Fetch three real URLs concurrently with one shared httpx.AsyncClient and asyncio.gather(), and measure the real elapsed time against what a sequential loop would take.

import asyncio
import time
import httpx

async def fetch_all(urls):
    async with httpx.AsyncClient() as client:
        start = time.monotonic()
        responses = await asyncio.gather(*(client.get(u) for u in urls))
        elapsed = time.monotonic() - start
    return [r.status_code for r in responses], elapsed

urls = [
    "https://httpbin.org/get?q=1",
    "https://httpbin.org/get?q=2",
    "https://httpbin.org/get?q=3",
]
statuses, elapsed = asyncio.run(fetch_all(urls))
print(statuses)
print(f"elapsed: {elapsed:.2f}s for {len(urls)} concurrent requests")

6

One AsyncClient is built and reused for all three requests inside the same async with block.

8

asyncio.gather() starts all three GET requests before awaiting any of their responses.

[200, 200, 200]
elapsed: 1.18s for 3 concurrent requests

Why this works: httpbin.org/get responds in well under a second per call, so three genuinely sequential requests would take roughly three times as long as one. Because asyncio.gather() starts all three before awaiting any response, the process spends most of its time waiting on the network concurrently rather than one request at a time — the real, measured payoff of an async client over a sync loop for independent I/O-bound calls.

What we're doing: Confirm raise_for_status() genuinely raises for a real 404 from both libraries, with the correct status code recoverable from the exception.

import requests
import httpx

r = requests.get('https://httpbin.org/status/404')
try:
    r.raise_for_status()
except requests.exceptions.HTTPError as e:
    print('requests HTTPError:', e.response.status_code, type(e).__name__)

r = httpx.get('https://httpbin.org/status/404')
try:
    r.raise_for_status()
except httpx.HTTPStatusError as e:
    print('httpx HTTPStatusError:', e.response.status_code, type(e).__name__)

4

httpbin.org/status/404 always returns a 404 with an empty body -- built specifically for testing status handling.

6

raise_for_status() inspects status_code itself and raises only because 404 is in the 4xx range.

requests HTTPError: 404 HTTPError
httpx HTTPStatusError: 404 HTTPStatusError

Why this works: Both libraries genuinely raise their respective exception type only after inspecting the real status code returned by the server, and both expose the original response (with its real status_code) on the caught exception — confirming raise_for_status() is a deliberate check, not something that happens automatically on every request.

What we're doing: Trigger a real JSONDecodeError by parsing an HTML response as JSON, then confirm a genuine JSON response parses and exposes its real keys.

import requests

# httpbin.org/html returns a real HTML page, not JSON
r = requests.get('https://httpbin.org/html')
try:
    r.json()
except requests.exceptions.JSONDecodeError as e:
    print('JSONDecodeError raised:', type(e).__name__)

# httpbin.org/json returns real, well-formed JSON
r = requests.get('https://httpbin.org/json')
data = r.json()
print('keys:', list(data.keys()))

4

/html deliberately returns text/html -- calling .json() on it must fail, since HTML is not valid JSON.

11

/json returns a real, well-formed JSON document -- .json() succeeds and the keys are inspectable normally.

JSONDecodeError raised: JSONDecodeError
keys: ['slideshow']

Why this works: The HTML response genuinely fails to parse as JSON — this is exactly the failure mode a misconfigured proxy or an API returning an error page produces in production, not a contrived edge case. The successful call confirms that even a well-formed JSON response has a specific shape (here, one top-level "slideshow" key) that calling code still has to know about and check, not just assume.

What we're doing: Show TLS verification actually rejecting a real self-signed certificate, proving the check is doing real work rather than being a formality.

import httpx

try:
    r = httpx.get("https://self-signed.badssl.com/", timeout=5)
    print("unexpected success:", r.status_code)
except httpx.ConnectError as e:
    print("rejected, as expected:", type(e).__name__)

4

self-signed.badssl.com deliberately serves a certificate no public CA has signed -- a live test target for verification failures.

6

httpx raises ConnectError (wrapping an SSL certificate verification failure) instead of silently connecting.

rejected, as expected: ConnectError

Why this works: httpx (like requests) verifies the server's certificate chain against a trusted CA bundle by default, and self-signed.badssl.com's certificate is deliberately signed by nobody a browser or HTTP client trusts. The connection fails BEFORE any HTTP request is sent, which is exactly the protection TLS verification exists to provide — proof the check is real, not a formality that could safely be skipped.