pip install requests

import requests

resp = requests.get("https://httpbin.org/json")
resp.raise_for_status()           # raises HTTPError for 4xx/5xx
data = resp.json()
print(resp.status_code)
print(data["slideshow"]["title"])

200
Sample Slide Show

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

session = requests.Session()
retry = Retry(total=3, backoff_factor=0.5, status_forcelist=[500, 502, 503, 504])
session.mount("https://", HTTPAdapter(max_retries=retry))

resp = session.post(
    "https://httpbin.org/post",
    json={"user": "alice", "action": "login"},
    headers={"X-App-Version": "1.0"},
    timeout=(3.05, 10),
)
resp.raise_for_status()
body = resp.json()
print(body["json"])
print(body["headers"]["Content-Type"])

{'action': 'login', 'user': 'alice'}
application/json

with requests.get("https://example.com/large.zip", stream=True, timeout=30) as r:
    r.raise_for_status()
    with open("large.zip", "wb") as f:
        for chunk in r.iter_content(chunk_size=8192):
            f.write(chunk)
print("Download complete")

Download complete

import requests
from requests.adapters import HTTPAdapter

session = requests.Session()

# Headers + auth that apply to every call on this session
session.headers.update({
    "User-Agent": "myapp/1.0",
    "Accept": "application/json",
})
session.auth = ("alicedev", "api-token")        # HTTP Basic on every call

# Tune connection pooling for high concurrency / many hosts
adapter = HTTPAdapter(pool_connections=20, pool_maxsize=50, pool_block=False)
session.mount("https://", adapter)
session.mount("http://", adapter)

# Per-call options override the session defaults
resp = session.get("https://api.example.com/users", params={"page": 1}, timeout=10)

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

retry_strategy = Retry(
    total=5,                                          # max retries (any reason)
    connect=3, read=3,                                # per-error-class caps
    status=5,                                         # retry on bad statuses
    backoff_factor=0.5,                               # 0.5, 1.0, 2.0, 4.0, …
    status_forcelist=(429, 500, 502, 503, 504),
    allowed_methods=("GET", "HEAD", "PUT", "DELETE", "OPTIONS", "POST"),
    respect_retry_after_header=True,                  # honor server Retry-After
    raise_on_status=False,                            # let raise_for_status() decide
)
adapter = HTTPAdapter(max_retries=retry_strategy)

session = requests.Session()
session.mount("https://", adapter)
session.mount("http://", adapter)

# Anything you call through `session` now retries automatically
resp = session.post("https://api.example.com/events", json={"event": "click"}, timeout=10)
resp.raise_for_status()

from requests.auth import HTTPBasicAuth, HTTPDigestAuth, AuthBase
import hmac, hashlib, time

# Built-ins
session.auth = HTTPBasicAuth("alicedev", "pw")
session.auth = HTTPDigestAuth("alicedev", "pw")

# Bearer token
class BearerAuth(AuthBase):
    def __init__(self, token: str):
        self.token = token
    def __call__(self, r):
        r.headers["Authorization"] = f"Bearer {self.token}"
        return r

# HMAC-signed requests
class HMACAuth(AuthBase):
    def __init__(self, key_id: str, secret: bytes):
        self.key_id, self.secret = key_id, secret
    def __call__(self, r):
        ts = str(int(time.time()))
        payload = (r.method + r.path_url + ts + (r.body or "")).encode()
        sig = hmac.new(self.secret, payload, hashlib.sha256).hexdigest()
        r.headers["X-Key-Id"] = self.key_id
        r.headers["X-Timestamp"] = ts
        r.headers["X-Signature"] = sig
        return r

session.auth = HMACAuth("kid_123", b"shared-secret")
resp = session.post("https://api.example.com/events", json={"event": "click"})

# Single file
with open("photo.jpg", "rb") as f:
    resp = session.post(
        "https://api.example.com/uploads",
        files={"photo": ("photo.jpg", f, "image/jpeg")},
        data={"caption": "Sunset"},   # form fields alongside the file
    )

# Multiple files (same field name)
files = [
    ("photos", ("a.jpg", open("a.jpg", "rb"), "image/jpeg")),
    ("photos", ("b.jpg", open("b.jpg", "rb"), "image/jpeg")),
]
session.post("https://api.example.com/album", files=files)

# True streaming multipart for huge files
# pip install requests-toolbelt
from requests_toolbelt.multipart.encoder import MultipartEncoder, MultipartEncoderMonitor

def show_progress(monitor):
    pct = monitor.bytes_read / monitor.len * 100
    print(f"\r{pct:5.1f}%  ({monitor.bytes_read:,}/{monitor.len:,})", end="")

encoder = MultipartEncoder(fields={
    "name": "big-video.mp4",
    "file": ("big-video.mp4", open("big-video.mp4", "rb"), "video/mp4"),
})
monitor = MultipartEncoderMonitor(encoder, show_progress)
session.post(
    "https://api.example.com/uploads",
    data=monitor,
    headers={"Content-Type": monitor.content_type},
)

import logging
import requests

log = logging.getLogger(__name__)

def log_response(resp, *args, **kwargs):
    log.info("%s %s -> %d %dms",
             resp.request.method, resp.url, resp.status_code,
             int(resp.elapsed.total_seconds() * 1000))

def assert_2xx(resp, *args, **kwargs):
    # Centralise raise_for_status() — every call gets it
    resp.raise_for_status()

session = requests.Session()
session.hooks["response"] = [log_response, assert_2xx]

resp = session.get("https://api.example.com/health", timeout=5)

# Tuple form: (connect timeout, read timeout) in seconds
requests.get(url, timeout=(3.05, 27))

# Both halves the same
requests.get(url, timeout=10)

# No timeout (BAD — only in scripts you'll babysit)
requests.get(url, timeout=None)

# Hard wall-clock budget
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FTimeout
def fetch():
    return requests.get("https://slow.example.com/", timeout=10)
with ThreadPoolExecutor(max_workers=1) as ex:
    try:
        resp = ex.submit(fetch).result(timeout=15)
    except FTimeout:
        log.error("Total budget exceeded")

# Custom CA bundle (e.g. corporate proxy issuing its own certs)
session.verify = "/etc/ssl/certs/internal-ca.pem"

# Mutual TLS (client cert + key)
session.cert = ("/etc/myapp/client.crt", "/etc/myapp/client.key")

# Pin to a specific protocol via a custom adapter (drops TLS < 1.2)
import ssl
from requests.adapters import HTTPAdapter
from urllib3 import PoolManager

class TLS12Adapter(HTTPAdapter):
    def init_poolmanager(self, *a, **kw):
        ctx = ssl.create_default_context()
        ctx.minimum_version = ssl.TLSVersion.TLSv1_2
        kw["ssl_context"] = ctx
        self.poolmanager = PoolManager(*a, **kw)
session.mount("https://", TLS12Adapter())

# Proxies (per-request or per-session)
session.proxies = {
    "http":  "http://proxy.internal:3128",
    "https": "http://proxy.internal:3128",
    # Bypass list:
    "no_proxy": "localhost,127.0.0.1,.internal.example.com",
}

# SOCKS5 proxy — pip install "requests[socks]"
session.proxies = {"https": "socks5h://localhost:1080"}

from requests.exceptions import (
    ConnectionError, ConnectTimeout, ReadTimeout, HTTPError, RequestException,
)

try:
    resp = session.get(url, timeout=(3, 10))
    resp.raise_for_status()
    return resp.json()
except ConnectTimeout:
    log.warning("Connect timeout — DNS/firewall issue")
except ReadTimeout:
    log.warning("Read timeout — slow server, will retry")
except HTTPError as e:
    log.error("HTTP %d: %s", e.response.status_code, e.response.text[:200])
except RequestException as e:
    log.exception("Unexpected request error: %s", e)

# Same call, three libraries
import requests
requests.get("https://httpbin.org/get").json()

import httpx
httpx.get("https://httpbin.org/get").json()

import aiohttp, asyncio
async def go():
    async with aiohttp.ClientSession() as s:
        async with s.get("https://httpbin.org/get") as r:
            return await r.json()
asyncio.run(go())

# Reusing the session matters — same-host TLS handshakes drop from ~200ms to ~0
session = requests.Session()
adapter = HTTPAdapter(pool_connections=20, pool_maxsize=100)
session.mount("https://", adapter)

# Parallel fan-out with a thread pool
from concurrent.futures import ThreadPoolExecutor, as_completed

def fetch(url: str) -> dict:
    r = session.get(url, timeout=10)
    r.raise_for_status()
    return r.json()

urls = [f"https://api.example.com/users/{i}" for i in range(100)]
with ThreadPoolExecutor(max_workers=20) as ex:
    futures = {ex.submit(fetch, u): u for u in urls}
    for fut in as_completed(futures):
        url = futures[fut]
        try:
            data = fut.result()
        except Exception as e:
            log.warning("Failed %s: %s", url, e)

import time, random
from requests.exceptions import RequestException

def poll(url: str, *, max_attempts: int = 10) -> dict:
    delay = 1.0
    for attempt in range(1, max_attempts + 1):
        try:
            r = session.get(url, timeout=10)
            if r.status_code == 200 and r.json().get("ready"):
                return r.json()
            if r.status_code in (429, 503):
                delay = float(r.headers.get("Retry-After", delay))
        except RequestException as e:
            log.warning("attempt %d failed: %s", attempt, e)
        # Exponential backoff with jitter
        time.sleep(delay + random.random())
        delay = min(delay * 2, 60)
    raise TimeoutError(f"{url} did not become ready in {max_attempts} attempts")

def all_users(session, url="https://api.example.com/users"):
    while url:
        r = session.get(url, params={"per_page": 100}, timeout=10)
        r.raise_for_status()
        yield from r.json()["items"]
        # GitHub-style Link header pagination
        url = r.links.get("next", {}).get("url")

import os

def download_with_resume(url: str, dest: str) -> None:
    head = session.head(url, timeout=10)
    total = int(head.headers.get("Content-Length", 0))
    pos = os.path.getsize(dest) if os.path.exists(dest) else 0
    if pos >= total > 0:
        return  # already complete
    headers = {"Range": f"bytes={pos}-"} if pos else {}
    with session.get(url, headers=headers, stream=True, timeout=30) as r:
        r.raise_for_status()
        with open(dest, "ab") as f:
            for chunk in r.iter_content(chunk_size=64 * 1024):
                f.write(chunk)

# pip install responses
import responses

@responses.activate
def test_create_user():
    responses.add(
        responses.POST,
        "https://api.example.com/users",
        json={"id": 1, "email": "alice@example.com"},
        status=201,
        match=[responses.matchers.json_params_matcher({"email": "alice@example.com"})],
    )
    user = create_user(email="alice@example.com")    # the code under test
    assert user["id"] == 1