Python Atlas
Standard library & async

HTTP with urllib

Build URLs and make dependency-free synchronous HTTP requests safely.

urllib.parse constructs and analyzes URLs. urllib.request is a synchronous HTTP client. It is sufficient for scripts and integrations with modest requirements; third-party clients can offer connection pooling, richer timeout controls, HTTP/2, and async APIs.

Encode URLs correctly

Do not concatenate untrusted query values:

from urllib.parse import urlencode, urljoin, urlsplit

query = urlencode({"q": "python async", "page": 2})
url = f"https://example.com/search?{query}"
parts = urlsplit(url)

assert parts.hostname == "example.com"

Use quote() for a path segment and urlencode() for query parameters. They use different escaping rules.

urljoin() treats an absolute second argument as a replacement URL:

from urllib.parse import urljoin

print(urljoin("https://trusted.example/users/", "//evil.example/path"))

Therefore, never use urljoin(base, user_input) as a security boundary. Parse, validate the scheme and normalized hostname, reject credentials, and apply an allowlist before connecting.

GET and response handling

import json
from urllib.request import Request, urlopen

request = Request(
    "https://api.example.com/status",
    headers={
        "Accept": "application/json",
        "User-Agent": "stdlib-course/1.0",
    },
)

with urlopen(request, timeout=10) as response:
    if response.status != 200:
        raise RuntimeError(f"unexpected HTTP status: {response.status}")
    body = response.read(1_000_001)
    if len(body) > 1_000_000:
        raise ValueError("response too large")
    payload = json.loads(body.decode(response.headers.get_content_charset() or "utf-8"))

A timeout is mandatory for network robustness, but urlopen exposes a single timeout rather than separate connect/read/pool limits. Always bound response size; a valid status does not imply safe content.

POST JSON

import json
from urllib.request import Request, urlopen

data = json.dumps({"name": "Ada"}).encode("utf-8")
request = Request(
    "https://api.example.com/users",
    data=data,
    method="POST",
    headers={
        "Content-Type": "application/json; charset=utf-8",
        "Accept": "application/json",
    },
)

with urlopen(request, timeout=10) as response:
    result = json.load(response)

json.load(response) expects JSON encoded in a compatible text form; explicit bounded reads are safer for untrusted responses.

Errors and retries

from urllib.error import HTTPError, URLError

try:
    with urlopen(request, timeout=10) as response:
        body = response.read()
except HTTPError as error:
    # HTTPError is also a response-like object.
    print(error.code, error.reason)
except URLError as error:
    print(f"network failure: {error.reason}")

Retry only transient failures, with capped exponential backoff and jitter. Retrying a non-idempotent POST can duplicate an operation unless the API supports idempotency keys. Honor Retry-After where applicable.

Security boundaries

Fetching user-supplied URLs creates server-side request forgery risk. Scheme and hostname checks alone are insufficient because DNS may resolve to loopback, private, link-local, or cloud metadata addresses, and redirects can change the destination. High-risk services need DNS/IP validation, redirect policy, outbound network controls, and response limits.

TLS verification is enabled by default. Do not disable certificate validation to fix a certificate error. Configure a correct trust store or server certificate.

urllib and asyncio

urlopen() blocks its thread. Calling it directly inside an event loop blocks every task:

import asyncio
from urllib.request import urlopen

def fetch(url: str) -> bytes:
    with urlopen(url, timeout=10) as response:
        return response.read(1_000_000)

async def async_fetch(url: str) -> bytes:
    return await asyncio.to_thread(fetch, url)

to_thread() is a bridge, not a fully async HTTP stack. Cancellation stops waiting for the thread's result but cannot forcibly terminate the underlying blocking call, so its own timeout is still essential.

Practice

Write a JSON fetcher that allows only HTTPS hosts from a fixed set, sends a user agent, limits responses to 512 KiB, distinguishes HTTP from transport errors, and retries 503 at most twice. Explain what additional protections are needed if redirects are enabled.

LEARNING RECORD

Finish this lesson

Mark it complete when you can explain the main decision without looking.

KNOWLEDGE CHECK

HTTP with urllib

1Why is urljoin("https://trusted.example/", user_input) insufficient as an SSRF security boundary?
2An async program calls urllib.request through asyncio.to_thread(). What remains true if the awaiting task is cancelled?