Calling REST APIs with requests

If you do DevOps in Python, you make HTTP calls — to cloud APIs, internal services, GitHub, Slack, monitoring vendors. The requests library is the canonical way; httpx is the modern replacement that adds async support with the same API.

Install

pip install requests

The Simple Case

import requests

resp = requests.get("https://api.github.com/repos/python/cpython", timeout=10)
resp.raise_for_status()        # raises if 4xx/5xx
data = resp.json()
print(data["stargazers_count"])

Three things you should always do:

1. Pass a timeout — without it, a hung server hangs your script forever
2. Call raise_for_status() or check resp.status_code — never assume success
3. Use .json() to parse JSON; it's safer than calling json.loads(resp.text)

POST, Headers, JSON Bodies

import os
import requests

token = os.environ["GITHUB_TOKEN"]
headers = {
    "Authorization": f"Bearer {token}",
    "Accept": "application/vnd.github+json",
}

payload = {"title": "Bug report", "body": "Something is broken"}

resp = requests.post(
    "https://api.github.com/repos/owner/repo/issues",
    headers=headers,
    json=payload,           # auto-encodes + sets Content-Type
    timeout=10,
)
resp.raise_for_status()
print(resp.json()["html_url"])

Use json=payload for JSON bodies, data=payload for form-encoded, and files={...} for multipart uploads.

Sessions

If you make more than one request to the same host, use a Session. It pools connections (significantly faster) and lets you set defaults once:

import os
import requests

session = requests.Session()
session.headers.update({
    "Authorization": f"Bearer {os.environ['GITHUB_TOKEN']}",
    "Accept": "application/vnd.github+json",
})

repos = session.get("https://api.github.com/user/repos", timeout=10).json()
issues = session.get("https://api.github.com/issues", timeout=10).json()

Query Parameters

resp = requests.get(
    "https://api.github.com/search/repositories",
    params={"q": "language:python", "sort": "stars", "per_page": 10},
    timeout=10,
)

requests URL-encodes parameters for you — never build query strings by hand.

Pagination

Most APIs paginate. Two common patterns:

Link header (GitHub, RFC 5988)

def fetch_all_issues(repo: str) -> list:
    issues = []
    url = f"https://api.github.com/repos/{repo}/issues?per_page=100"
    while url:
        resp = session.get(url, timeout=10)
        resp.raise_for_status()
        issues.extend(resp.json())
        # requests parses Link: <...>; rel="next" automatically
        url = resp.links.get("next", {}).get("url")
    return issues

Cursor / offset

items = []
cursor = None
while True:
    params = {"limit": 100}
    if cursor:
        params["cursor"] = cursor
    resp = session.get(api_url, params=params, timeout=10)
    resp.raise_for_status()
    page = resp.json()
    items.extend(page["items"])
    cursor = page.get("next_cursor")
    if not cursor:
        break

Retries with Backoff

Networks fail. Retry transient errors with exponential backoff. The cleanest way is the built-in adapter:

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

retry = Retry(
    total=5,
    backoff_factor=1,                 # 1s, 2s, 4s, 8s, 16s
    status_forcelist=[429, 500, 502, 503, 504],
    allowed_methods=["GET", "POST", "PUT", "DELETE"],
)

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

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

Be careful retrying non-idempotent calls — POST that creates a resource may end up creating it twice. Use idempotency keys when the API supports them.

Error Handling

try:
    resp = session.get(url, timeout=10)
    resp.raise_for_status()
except requests.Timeout:
    print("timed out")
except requests.ConnectionError:
    print("could not connect")
except requests.HTTPError as e:
    print(f"http error: {e.response.status_code} {e.response.text}")
except requests.RequestException as e:
    print(f"other request error: {e}")

Streaming Large Responses

with session.get("https://example.com/big.tar.gz", stream=True, timeout=30) as resp:
    resp.raise_for_status()
    with open("big.tar.gz", "wb") as f:
        for chunk in resp.iter_content(chunk_size=8192):
            f.write(chunk)

httpx: The Async-Capable Cousin

# pip install httpx
import httpx

with httpx.Client(timeout=10) as client:
    resp = client.get("https://api.github.com")
    print(resp.json())

# async version — see the async lesson
import asyncio

async def main():
    async with httpx.AsyncClient(timeout=10) as client:
        resp = await client.get("https://api.github.com")
        return resp.json()

asyncio.run(main())

Same API, but you can call many endpoints concurrently — covered in the async lesson.

Debugging Requests

When an API call misbehaves, inspect the wire:

resp = session.get(url, params={"q": "test"}, timeout=10)
print(resp.url)              # final URL with encoded params
print(resp.status_code)
print(resp.headers)
print(resp.text[:500])       # first 500 chars of response body

For deeper inspection, set the requests logger to DEBUG, or run your script through mitmproxy for full request/response capture.