Working with APIs

# Basic GET request — fetching data from a public API

import requests

# Fetch a post from JSONPlaceholder (a free fake REST API for testing)
url = "https://jsonplaceholder.typicode.com/posts/1"
response = requests.get(url)

# Status code — 200 means OK
print("Status:", response.status_code)
print("Content-Type:", response.headers["Content-Type"])

# Parse the JSON body
data = response.json()
print("Title:", data["title"])
print("User ID:", data["userId"])

# Query parameters — appended to the URL as ?key=value
response = requests.get(
    "https://jsonplaceholder.typicode.com/posts",
    params={"userId": 1, "_limit": 3}   # fetch first 3 posts by user 1
)
posts = response.json()
print(f"\nFetched {len(posts)} posts:")
for post in posts:
    print(f"  [{post['id']}] {post['title'][:40]}...")

# POST, PUT, PATCH, DELETE — write operations

import requests

BASE = "https://jsonplaceholder.typicode.com"

# POST — create a new resource
new_post = {"title": "My New Post", "body": "Hello world", "userId": 1}
r = requests.post(f"{BASE}/posts", json=new_post)
print("POST status:", r.status_code)    # 201 Created
print("Created:", r.json())

# PUT — replace an existing resource entirely
updated = {"title": "Updated Title", "body": "New body", "userId": 1}
r = requests.put(f"{BASE}/posts/1", json=updated)
print("\nPUT status:", r.status_code)   # 200 OK

# PATCH — partially update a resource
r = requests.patch(f"{BASE}/posts/1", json={"title": "Patched Title"})
print("PATCH title:", r.json()["title"])

# DELETE — remove a resource
r = requests.delete(f"{BASE}/posts/1")
print("DELETE status:", r.status_code)  # 200 OK

# Robust error handling for API requests

import requests
from requests.exceptions import RequestException, Timeout, ConnectionError

def safe_get(url, params=None, timeout=5):
    """Make a GET request with comprehensive error handling."""
    try:
        response = requests.get(url, params=params, timeout=timeout)
        response.raise_for_status()   # raises HTTPError for 4xx and 5xx
        return response.json()

    except Timeout:
        print(f"Request timed out after {timeout}s: {url}")
    except ConnectionError:
        print(f"Could not connect to: {url}")
    except requests.exceptions.HTTPError as e:
        code = e.response.status_code
        if code == 401:
            print("Unauthorised — check your API key")
        elif code == 404:
            print(f"Resource not found: {url}")
        elif code == 429:
            print("Rate limit hit — slow down requests")
        else:
            print(f"HTTP error {code}: {e}")
    except RequestException as e:
        print(f"Unexpected request error: {e}")
    return None

# Test with a valid URL
data = safe_get("https://jsonplaceholder.typicode.com/users/1")
if data:
    print(f"User: {data['name']} — {data['email']}")

# Test with a 404
safe_get("https://jsonplaceholder.typicode.com/posts/99999")

# Authentication patterns

import requests

# 1. API Key in query string
requests.get(
    "https://api.example.com/data",
    params={"api_key": "your_key_here"}
)

# 2. API Key in header (most common)
headers = {"X-API-Key": "your_key_here"}
requests.get("https://api.example.com/data", headers=headers)

# 3. Bearer token (OAuth2 / JWT)
token = "eyJhbGciOiJIUzI1NiIs..."
headers = {"Authorization": f"Bearer {token}"}
requests.get("https://api.example.com/protected", headers=headers)

# 4. Basic Auth — username and password
requests.get(
    "https://api.example.com/resource",
    auth=("username", "password")
)

# Best practice — load secrets from environment variables, never hardcode
import os
api_key = os.environ.get("API_KEY")
headers = {"X-API-Key": api_key}
print("Using API key from environment:", bool(api_key))

# requests.Session — reuse connection and headers

import requests

BASE = "https://jsonplaceholder.typicode.com"

with requests.Session() as session:
    # Set headers once — applied to every request made through this session
    session.headers.update({
        "Authorization": "Bearer fake-token-123",
        "Accept": "application/json"
    })

    # All requests reuse the same connection and headers
    user    = session.get(f"{BASE}/users/1").json()
    posts   = session.get(f"{BASE}/posts", params={"userId": 1}).json()
    todos   = session.get(f"{BASE}/todos", params={"userId": 1}).json()

print(f"User: {user['name']}")
print(f"Posts: {len(posts)}")
print(f"Todos: {len(todos)}")

# Pagination — fetch all pages of results

import requests

def fetch_all_posts():
    """Fetch all posts using page-based pagination."""
    all_posts = []
    page      = 1

    while True:
        response = requests.get(
            "https://jsonplaceholder.typicode.com/posts",
            params={"_page": page, "_limit": 10}
        )
        data = response.json()

        if not data:          # empty list means no more pages
            break

        all_posts.extend(data)
        print(f"Page {page}: fetched {len(data)} posts (total so far: {len(all_posts)})")
        page += 1

    return all_posts

posts = fetch_all_posts()
print(f"\nTotal posts fetched: {len(posts)}")

# Reusable API client class

import requests
from requests.exceptions import RequestException

class APIClient:
    def __init__(self, base_url, api_key=None):
        self.base_url = base_url.rstrip("/")
        self.session  = requests.Session()
        if api_key:
            self.session.headers.update({"X-API-Key": api_key})
        self.session.headers.update({"Accept": "application/json"})

    def get(self, endpoint, **kwargs):
        return self._request("GET", endpoint, **kwargs)

    def post(self, endpoint, **kwargs):
        return self._request("POST", endpoint, **kwargs)

    def _request(self, method, endpoint, **kwargs):
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        try:
            r = self.session.request(method, url, timeout=10, **kwargs)
            r.raise_for_status()
            return r.json()
        except RequestException as e:
            print(f"[{method}] {url} failed: {e}")
            return None

    def close(self):
        self.session.close()

# Usage
client = APIClient("https://jsonplaceholder.typicode.com")
user   = client.get("/users/1")
posts  = client.get("/posts", params={"userId": 1, "_limit": 3})

if user:
    print(f"User: {user['name']}")
if posts:
    print(f"Posts returned: {len(posts)}")
client.close()