HTTP Requests with the requests Library

# pip install requests
import requests

# Simple GET request
response = requests.get("https://api.github.com/users/python")

print(response.status_code)    # 200
print(response.headers['Content-Type'])  # application/json; charset=utf-8

# Parse JSON response
data = response.json()
print(data['name'])            # Python
print(data['public_repos'])    # Number of repos

# Check if request succeeded
if response.ok:  # True for 200-299 status codes
    print("Success!")

response.raise_for_status()  # Raises HTTPError for 4xx/5xx responses

# Search GitHub repositories
params = {
    "q": "machine learning language:python",
    "sort": "stars",
    "order": "desc",
    "per_page": 10
}

response = requests.get(
    "https://api.github.com/search/repositories",
    params=params
)

# requests builds: ?q=machine+learning+language%3Apython&sort=stars&...
data = response.json()
for repo in data['items'][:3]:
    print(f"{repo['full_name']}: {repo['stargazers_count']:,} stars")

import json

# POST: Create a resource
new_post = {
    "title": "My First Post",
    "body": "Hello, World!",
    "userId": 1
}

response = requests.post(
    "https://jsonplaceholder.typicode.com/posts",
    json=new_post  # Automatically sets Content-Type: application/json
)
print(response.status_code)  # 201 Created
print(response.json())

# PUT: Replace a resource
updated_post = {"title": "Updated Post", "body": "Updated content", "userId": 1}
response = requests.put(
    "https://jsonplaceholder.typicode.com/posts/1",
    json=updated_post
)

# PATCH: Partial update
response = requests.patch(
    "https://jsonplaceholder.typicode.com/posts/1",
    json={"title": "New Title"}
)

# DELETE: Remove a resource
response = requests.delete("https://jsonplaceholder.typicode.com/posts/1")
print(response.status_code)  # 200

# API Key in headers
headers = {"Authorization": "Bearer your-api-token-here"}
response = requests.get("https://api.example.com/data", headers=headers)

# API Key in query params
response = requests.get("https://api.example.com/data", params={"api_key": "your-key"})

# HTTP Basic Auth
response = requests.get(
    "https://api.github.com/user",
    auth=("username", "personal-access-token")
)

# OAuth2 Bearer Token
import os
token = os.getenv("GITHUB_TOKEN")
headers = {
    "Authorization": f"Bearer {token}",
    "Accept": "application/vnd.github.v3+json"
}
response = requests.get("https://api.github.com/user", headers=headers)

# Session: reuses TCP connections and persists headers/cookies
session = requests.Session()

# Set headers once for all requests
session.headers.update({
    "Authorization": f"Bearer {token}",
    "User-Agent": "MyApp/1.0"
})

# All requests use the session headers
r1 = session.get("https://api.example.com/users")
r2 = session.get("https://api.example.com/posts")

session.close()

# Better: use as context manager
with requests.Session() as session:
    session.headers["Authorization"] = f"Bearer {token}"
    users = session.get("https://api.example.com/users").json()
    posts = session.get("https://api.example.com/posts").json()

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

def api_request(url, max_retries=3, timeout=10):
    """Make an HTTP request with retry logic."""
    for attempt in range(max_retries):
        try:
            response = requests.get(url, timeout=timeout)
            response.raise_for_status()  # Raise for 4xx/5xx
            return response.json()
        
        except ConnectionError:
            print(f"Connection failed (attempt {attempt+1})")
        except Timeout:
            print(f"Request timed out (attempt {attempt+1})")
        except HTTPError as e:
            if e.response.status_code == 429:  # Rate limited
                retry_after = int(e.response.headers.get('Retry-After', 60))
                print(f"Rate limited. Waiting {retry_after}s...")
                time.sleep(retry_after)
            elif e.response.status_code < 500:
                raise  # Client errors (4xx) don't benefit from retrying
            else:
                print(f"Server error {e.response.status_code} (attempt {attempt+1})")
        
        if attempt < max_retries - 1:
            wait = 2 ** attempt  # Exponential backoff: 1s, 2s, 4s
            time.sleep(wait)
    
    raise RequestException(f"Failed after {max_retries} attempts")

from pathlib import Path

def download_file(url, destination, chunk_size=8192):
    """Download a file with progress display."""
    response = requests.get(url, stream=True)
    response.raise_for_status()
    
    total_size = int(response.headers.get('Content-Length', 0))
    downloaded = 0
    
    dest_path = Path(destination)
    dest_path.parent.mkdir(parents=True, exist_ok=True)
    
    with open(dest_path, 'wb') as f:
        for chunk in response.iter_content(chunk_size=chunk_size):
            f.write(chunk)
            downloaded += len(chunk)
            if total_size:
                pct = downloaded / total_size * 100
                print(f"\r{pct:.1f}%", end='', flush=True)
    
    print(f"\nDownloaded: {dest_path}")

download_file("https://example.com/large-dataset.csv", "data/dataset.csv")

class GitHubClient:
    """A simple GitHub API client."""
    
    BASE_URL = "https://api.github.com"
    
    def __init__(self, token=None):
        self.session = requests.Session()
        self.session.headers.update({
            "Accept": "application/vnd.github.v3+json",
            "User-Agent": "MyApp/1.0"
        })
        if token:
            self.session.headers["Authorization"] = f"Bearer {token}"
    
    def _request(self, method, path, **kwargs):
        url = f"{self.BASE_URL}{path}"
        response = self.session.request(method, url, **kwargs)
        response.raise_for_status()
        return response.json()
    
    def get_user(self, username):
        return self._request("GET", f"/users/{username}")
    
    def get_repos(self, username, sort="updated"):
        return self._request("GET", f"/users/{username}/repos", 
                             params={"sort": sort})
    
    def search_repos(self, query, language=None, min_stars=0):
        q = query
        if language:
            q += f" language:{language}"
        if min_stars:
            q += f" stars:>={min_stars}"
        return self._request("GET", "/search/repositories", 
                            params={"q": q, "sort": "stars"})
    
    def __enter__(self):
        return self
    
    def __exit__(self, *args):
        self.session.close()

# Usage
with GitHubClient(token=os.getenv("GITHUB_TOKEN")) as github:
    user = github.get_user("python")
    repos = github.get_repos("python")
    results = github.search_repos("machine learning", language="python", min_stars=1000)
    
    print(f"User: {user['name']} ({user['public_repos']} repos)")
    for repo in results['items'][:5]:
        print(f"  {repo['full_name']}: {repo['stargazers_count']:,} stars")