HTTP Requests

Master REST API calls, session management, and authentication patterns using requests and httpx.

---

Installation

pip install requests httpx

---

requests — Basic HTTP Client

import requests

# GET
response = requests.get("https://api.github.com/users/octocat")
print(response.status_code)   # 200
print(response.json())        # dict
print(response.headers["content-type"])

# POST
data = {"username": "alice", "password": "secret"}
response = requests.post("https://httpbin.org/post", json=data)

# PUT / PATCH / DELETE
requests.put("https://api.example.com/items/1", json={"name": "Updated"})
requests.patch("https://api.example.com/items/1", json={"stock": 5})
requests.delete("https://api.example.com/items/1")

---

Request Parameters

# Query parameters
response = requests.get(
    "https://api.example.com/products",
    params={"page": 1, "limit": 20, "category": "books"},
)
# → https://api.example.com/products?page=1&limit=20&category=books

# Headers
headers = {
    "Authorization": "Bearer eyJhbGciOiJIUzI1NiJ9...",
    "Content-Type": "application/json",
    "User-Agent": "MyApp/1.0",
}
response = requests.get("https://api.example.com/me", headers=headers)

# Timeout (always set this!)
response = requests.get("https://api.example.com/data",
                        timeout=(5, 30))  # (connect timeout, read timeout)

# File upload
with open("photo.jpg", "rb") as f:
    response = requests.post(
        "https://api.example.com/upload",
        files={"file": ("photo.jpg", f, "image/jpeg")},
        data={"description": "Profile photo"},
    )

---

Session — Connection Reuse + Shared Config

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

# Retry strategy
retry_strategy = Retry(
    total=3,               # max 3 retries
    backoff_factor=1,      # 1s, 2s, 4s exponential backoff
    status_forcelist=[429, 500, 502, 503, 504],
)
adapter = HTTPAdapter(max_retries=retry_strategy)

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

# Shared headers
session.headers.update({
    "Authorization": "Bearer my_token",
    "Accept": "application/json",
})

# Requests using session (reuses connection, preserves cookies)
r1 = session.get("https://api.example.com/users")
r2 = session.get("https://api.example.com/products")

session.close()  # or use context manager
with requests.Session() as s:
    s.headers.update({"Authorization": "Bearer token"})
    response = s.get("https://api.example.com/data")

---

Error Handling

import requests
from requests.exceptions import (
    Timeout,
    ConnectionError,
    HTTPError,
    RequestException,
)


def safe_request(url: str, **kwargs) -> dict | None:
    try:
        response = requests.get(url, timeout=10, **kwargs)
        response.raise_for_status()  # 4xx, 5xx → raises HTTPError
        return response.json()
    except Timeout:
        print(f"Request timed out: {url}")
    except ConnectionError:
        print(f"Connection failed: {url}")
    except HTTPError as e:
        print(f"HTTP error {e.response.status_code}: {url}")
        if e.response.status_code == 401:
            print("Authentication required — check your token")
        elif e.response.status_code == 429:
            retry_after = e.response.headers.get("Retry-After", 60)
            print(f"Retry after {retry_after}s")
    except RequestException as e:
        print(f"Request failed: {e}")
    return None

---

httpx — Sync/Async Unified Client

import httpx

# Sync (similar API to requests)
with httpx.Client(timeout=10.0, follow_redirects=True) as client:
    response = client.get("https://api.github.com/users/octocat")
    data = response.json()

# Async (asyncio-based)
import asyncio


async def fetch_multiple(urls: list[str]) -> list[dict]:
    async with httpx.AsyncClient(timeout=10.0) as client:
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)
        return [r.json() for r in responses if r.status_code == 200]


# Parallel API calls
async def get_user_data(user_ids: list[int]) -> list[dict]:
    async with httpx.AsyncClient(
        base_url="https://api.example.com",
        headers={"Authorization": "Bearer token"},
        timeout=15.0,
    ) as client:
        tasks = [client.get(f"/users/{uid}") for uid in user_ids]
        responses = await asyncio.gather(*tasks, return_exceptions=True)

        results = []
        for uid, resp in zip(user_ids, responses):
            if isinstance(resp, Exception):
                print(f"User {uid} fetch failed: {resp}")
            elif resp.status_code == 200:
                results.append(resp.json())
        return results


# Run
user_ids = [1, 2, 3, 4, 5]
users = asyncio.run(get_user_data(user_ids))

---

Authentication Patterns

import requests
from requests.auth import HTTPBasicAuth, HTTPDigestAuth


# 1. Basic Auth
response = requests.get(url, auth=HTTPBasicAuth("user", "pass"))
response = requests.get(url, auth=("user", "pass"))  # shorthand

# 2. Bearer Token (JWT)
headers = {"Authorization": f"Bearer {access_token}"}
response = requests.get(url, headers=headers)

# 3. API Key
response = requests.get(url, headers={"X-API-Key": "my_api_key"})
response = requests.get(url, params={"api_key": "my_api_key"})

# 4. OAuth2 (automatic token refresh)
from requests_oauthlib import OAuth2Session  # pip install requests-oauthlib

oauth = OAuth2Session(client_id, token=token)
response = oauth.get("https://api.example.com/resource")

---

REST API Client Class Pattern

import requests
from typing import Any


class APIClient:
    def __init__(self, base_url: str, api_key: str):
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        })
        self.base_url = base_url.rstrip("/")

    def _request(self, method: str, path: str, **kwargs) -> Any:
        url = f"{self.base_url}/{path.lstrip('/')}"
        response = self.session.request(method, url, timeout=15, **kwargs)
        response.raise_for_status()
        return response.json() if response.content else None

    def get(self, path: str, params: dict = None): return self._request("GET", path, params=params)
    def post(self, path: str, data: dict): return self._request("POST", path, json=data)
    def put(self, path: str, data: dict): return self._request("PUT", path, json=data)
    def delete(self, path: str): return self._request("DELETE", path)

    def __enter__(self): return self
    def __exit__(self, *args): self.session.close()


# Usage
with APIClient("https://api.example.com", api_key="abc123") as client:
    users = client.get("/users", params={"page": 1})
    new_user = client.post("/users", data={"name": "Alice"})

---

Summary

Feature	requests	httpx
Sync requests	✅	✅
Async requests	❌	✅ (AsyncClient)
HTTP/2	❌	✅
Type hints	partial	✅
Retries	HTTPAdapter	httpx.HTTPTransport

Use requests for simple synchronous calls, httpx when you need async or HTTP/2 support.