Python Requests Library Explained — HTTP, APIs and Real-World Patterns

import requests

# GitHub's public API — no auth required for basic profile lookups
# This is a real endpoint you can run right now
GITHUB_API_URL = "https://api.github.com/users/torvalds"

# Make the GET request — Requests handles TCP connection setup,
# HTTP header negotiation, response buffering, and decoding
response = requests.get(
    url=GITHUB_API_URL,
    timeout=(3, 10)  # 3s connect timeout, 10s read timeout — always set this
)

# raise_for_status() raises requests.exceptions.HTTPError for any 4xx or 5xx
# Call this before you touch the response body — always
response.raise_for_status()

# .json() parses the JSON body into a Python dict automatically
# No need for json.loads(response.text) — that's the old way
user_profile = response.json()

print(f"Name        : {user_profile['name']}")
print(f"Public Repos: {user_profile['public_repos']}")
print(f"Followers   : {user_profile['followers']}")
print(f"Profile URL : {user_profile['html_url']}")

# The Response object gives you full access to the HTTP layer
print(f"\nStatus Code : {response.status_code}")
print(f"Content-Type: {response.headers['Content-Type']}")
print(f"Rate Limit  : {response.headers.get('X-RateLimit-Remaining', 'N/A')} requests remaining")

import requests
import os

# Load credentials from environment — never hardcode tokens in source code
github_token = os.environ.get("GITHUB_TOKEN")

if not github_token:
    raise EnvironmentError(
        "GITHUB_TOKEN environment variable is not set. "
        "Generate a personal access token at https://github.com/settings/tokens"
    )

REPO_OWNER = "your-username"
REPO_NAME  = "your-test-repo"
API_URL    = f"https://api.github.com/repos/{REPO_OWNER}/{REPO_NAME}/issues"

# The payload we are sending — a plain Python dict
# Requests will serialise this to JSON and set the Content-Type header automatically
issue_payload = {
    "title": "Bug: Login page crashes on empty password field",
    "body": (
        "Steps to reproduce:\n"
        "1. Navigate to /login\n"
        "2. Leave the password field blank\n"
        "3. Click Submit\n\n"
        "Expected: Client-side validation error\n"
        "Actual: 500 Internal Server Error — unhandled NullPointerException in auth middleware"
    ),
    "labels": ["bug", "high-priority"]
}

# Headers set once — GitHub requires Accept and the API version header
request_headers = {
    "Authorization": f"Bearer {github_token}",
    "Accept": "application/vnd.github+json",
    "X-GitHub-Api-Version": "2022-11-28"
}

# json=issue_payload is correct — NOT data=json.dumps(issue_payload)
# The difference: json= sets Content-Type to application/json automatically
# data= would set Content-Type to application/x-www-form-urlencoded — the server would reject it
response = requests.post(
    url=API_URL,
    json=issue_payload,
    headers=request_headers,
    timeout=(3, 10)
)

# 201 Created is the correct success code for resource creation — not 200
# raise_for_status() handles anything that's not 2xx
response.raise_for_status()

created_issue = response.json()
print(f"Issue #{created_issue['number']} created successfully.")
print(f"Title : {created_issue['title']}")
print(f"URL   : {created_issue['html_url']}")
print(f"State : {created_issue['state']}")

# Demonstrate query parameters — fetch only open issues, 5 per page
# params= builds the URL query string: ...?state=open&per_page=5
# Never build this string manually — Requests handles encoding edge cases correctly
search_params = {"state": "open", "per_page": 5, "sort": "created", "direction": "desc"}

open_issues_response = requests.get(
    url=API_URL,
    params=search_params,
    headers=request_headers,
    timeout=(3, 10)
)
open_issues_response.raise_for_status()

open_issues = open_issues_response.json()
print(f"\nMost recent open issues ({len(open_issues)} returned):")
for issue in open_issues:
    print(f"  #{issue['number']}: {issue['title']}")

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


def build_resilient_session(
    api_token: str,
    pool_connections: int = 10,
    pool_maxsize: int = 50,
) -> requests.Session:
    """
    Creates a production-grade requests Session with:
      - Shared auth headers set once, inherited by all requests
      - Connection pooling to reuse TCP connections across calls
      - Automatic retry on transient failures with exponential backoff
      - Explicit pool size limits to prevent connection exhaustion

    Args:
        api_token:        Bearer token for Authorization header
        pool_connections: Number of distinct host connection pools to maintain
        pool_maxsize:     Max simultaneous connections per pool

    Returns:
        Configured requests.Session ready for production use
    """
    session = requests.Session()

    # Headers set here are sent with every request through this session
    # No need to repeat them on individual calls
    session.headers.update({
        "Authorization": f"Bearer {api_token}",
        "Accept": "application/json",
        "Content-Type": "application/json",
        "User-Agent": "TheCodeForge-Client/1.0"
    })

    # Retry strategy: retryable status codes are transient infrastructure failures
    # Do NOT add 400, 401, 403, 404 — those are permanent and will always fail the same way
    # backoff_factor=1 produces waits of: 0s, 2s, 4s, 8s between successive retries
    # (formula: backoff_factor * (2 ** (retry_number - 1)))
    retry_strategy = Retry(
        total=3,
        backoff_factor=1,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["GET", "HEAD", "OPTIONS"],  # Only retry idempotent methods by default
        raise_on_status=False   # Let raise_for_status() handle this — do not raise inside retry
    )

    # HTTPAdapter wraps the retry strategy and sets connection pool limits
    # pool_maxsize prevents connection exhaustion under concurrent load
    http_adapter = HTTPAdapter(
        max_retries=retry_strategy,
        pool_connections=pool_connections,
        pool_maxsize=pool_maxsize
    )

    # Mount the adapter for both HTTPS and HTTP — always cover both prefixes
    session.mount("https://", http_adapter)
    session.mount("http://",  http_adapter)

    return session


# --- Usage ---
api_token = os.environ.get("WEATHER_API_KEY", "demo_key")
BASE_URL  = "https://api.open-meteo.com/v1"

session = build_resilient_session(api_token=api_token)

try:
    response = session.get(
        url=f"{BASE_URL}/forecast",
        params={
            "latitude": 51.5074,        # London
            "longitude": -0.1278,
            "current_weather": True,
            "wind_speed_unit": "kmh"
        },
        timeout=(3, 10)                 # Always set timeout even on sessions
        # Note: timeout is NOT persisted on the session itself — set it per-call
    )
    response.raise_for_status()

    weather_data = response.json()
    current      = weather_data["current_weather"]

    print(f"Location    : London, UK")
    print(f"Temperature : {current['temperature']} °C")
    print(f"Wind Speed  : {current['windspeed']} km/h")
    print(f"Wind Dir    : {current['winddirection']}°")
    print(f"Weather Code: {current['weathercode']}")
    print(f"\nResponse time: {response.elapsed.total_seconds():.3f}s")

except requests.exceptions.Timeout:
    # Raised when connect_timeout or read_timeout is exceeded
    # Do not retry here — the retry strategy handles that inside the session
    print("ERROR: The weather API did not respond in time. Try again in a few seconds.")

except requests.exceptions.ConnectionError:
    # DNS failure, refused connection, or network unreachable
    print("ERROR: Could not reach the weather API. Check network connectivity.")

except requests.exceptions.HTTPError as http_err:
    # Raised by raise_for_status() for 4xx and 5xx responses that exhausted retries
    print(f"ERROR: HTTP {http_err.response.status_code} — {http_err}")

finally:
    # Always close the session to release the underlying connection pool
    # Use as a context manager (with build_resilient_session(...) as session:) to handle this automatically
    session.close()

import requests
from requests.auth import HTTPBasicAuth
import os


# ─── PATTERN 1: Basic Auth ────────────────────────────────────────────────────
# Requests base64-encodes "username:password" and sets the Authorization header
# Use HTTPBasicAuth explicitly for clarity — the tuple shorthand also works:
# auth=("demo_user", "secret123") is identical to HTTPBasicAuth(...)

print("=== Pattern 1: Basic Auth ===")
basic_response = requests.get(
    url="https://httpbin.org/basic-auth/demo_user/secret123",
    auth=HTTPBasicAuth("demo_user", "secret123"),
    timeout=(3, 10)
)
basic_response.raise_for_status()
result = basic_response.json()
print(f"Authenticated : {result['authenticated']}")
print(f"User          : {result['user']}")
print(f"Auth header   : {basic_response.request.headers.get('Authorization')}")
# Output will show: Basic ZGVtb191c2VyOnNlY3JldDEyMw== (base64 of demo_user:secret123)


# ─── PATTERN 2: Bearer Token Auth via Session ─────────────────────────────────
# Set the token once on the session — every subsequent call inherits it
# No need to pass headers= on every individual request

print("\n=== Pattern 2: Bearer Token via Session ===")
api_token = os.environ.get("MY_API_TOKEN", "mock_token_for_demo_abc123")

token_session = requests.Session()
token_session.headers.update({
    "Authorization": f"Bearer {api_token}",
    "Accept": "application/json"
})

try:
    token_response = token_session.get(
        url="https://httpbin.org/bearer",
        timeout=(3, 10)
    )
    token_response.raise_for_status()
    result = token_response.json()
    print(f"Authenticated : {result['authenticated']}")
    print(f"Token received: {result['token']}")
finally:
    token_session.close()  # Release the connection pool


# ─── PATTERN 3: API Key as Custom Header ──────────────────────────────────────
# Some APIs (OpenAI uses Authorization: Bearer, Stripe uses Authorization: Bearer,
# but others use custom header names — always check the docs)
# Header-based keys are safer than query param keys — they stay out of server logs

print("\n=== Pattern 3: API Key as Custom Header ===")
api_key = os.environ.get("SERVICE_API_KEY", "sk-demo-key-for-testing-9999")

header_key_response = requests.get(
    url="https://httpbin.org/headers",
    headers={
        "X-API-Key": api_key,
        "Accept": "application/json"
    },
    timeout=(3, 10)
)
header_key_response.raise_for_status()

returned_headers = header_key_response.json()["headers"]
print(f"API Key received: {returned_headers.get('X-Api-Key', 'not found')}")
print("Key delivered via header — does not appear in URL or server access logs")


# ─── PATTERN 4: API Key as Query Parameter (avoid if header-based is available) ─
# Some older APIs only support query param auth — use it only if you have no choice
# Never log the full URL when using query-param API keys

print("\n=== Pattern 4: API Key as Query Param (legacy — avoid when possible) ===")
query_key_response = requests.get(
    url="https://httpbin.org/get",
    params={"api_key": api_key, "other_param": "value"},
    timeout=(3, 10)
)
query_key_response.raise_for_status()
print(f"Full URL (key visible): {query_key_response.url}")
print("WARNING: This URL — and the key in it — will appear in every server access log")