Query REST APIs, handle authentication, parse responses, and build reliable integrations with cloud services—the bridge between your scripts and the infrastructure you manage.
An API is like ordering from a restaurant: you send a request ("I want 2 pizzas"), and they send back a response ("Here is your order"). In programming, you use HTTP to send requests to servers (GET, POST, PUT, DELETE), and they respond with data (usually JSON). The requests library makes this easy—it handles the low-level details so you just write request/response logic.
HTTP methods: GET (read), POST (create), PUT (replace), PATCH (update), DELETE (destroy). Status codes: 200 (success), 4xx (client error), 5xx (server error). Headers: metadata (auth tokens, content-type). Body: request data (usually JSON) or response data.
API keys, passwords, and tokens belong in environment variables or secure vaults—never in source code. Use os.getenv("API_KEY") to read from environment. Cloud providers (Azure Key Vault, AWS Secrets Manager) have client libraries for secure access.
import requests
import json
import time
from requests.auth import HTTPBasicAuth
# ===== SIMPLE GET REQUEST =====
response = requests.get("https://api.example.com/users")
status_code = response.status_code
if status_code == 200:
data = response.json() # parse response as JSON
print(f"Got {len(data)} users")
else:
print(f"Error: {status_code}")
# ===== HEADERS AND AUTHENTICATION =====
# Bearer token (common for RESTful APIs)
headers = {
"Authorization": f"Bearer {api_token}",
"Content-Type": "application/json",
"User-Agent": "MyScript/1.0"
}
response = requests.get("https://api.example.com/resources", headers=headers)
# Basic authentication (username:password encoded in base64)
response = requests.get(
"https://api.example.com/data",
auth=HTTPBasicAuth("user", "password")
)
# ===== POST REQUEST (CREATE DATA) =====
new_user = {
"name": "Alice",
"email": "alice@example.com",
"role": "admin"
}
response = requests.post(
"https://api.example.com/users",
json=new_user, # automatically serializes to JSON
headers=headers
)
if response.status_code == 201: # Created
created_user = response.json()
print(f"User created with ID: {created_user['id']}")
else:
print(f"Failed: {response.status_code}")
print(response.text) # raw response body for debugging
# ===== PUT REQUEST (REPLACE) =====
updated_user = {
"name": "Alice Smith",
"email": "alice.smith@example.com"
}
response = requests.put(
"https://api.example.com/users/123",
json=updated_user,
headers=headers
)
# ===== PATCH REQUEST (PARTIAL UPDATE) =====
# Update just one field
response = requests.patch(
"https://api.example.com/users/123",
json={"role": "viewer"},
headers=headers
)
# ===== DELETE REQUEST =====
response = requests.delete(
"https://api.example.com/users/123",
headers=headers
)
if response.status_code == 204: # No Content
print("User deleted")
# ===== PARAMETERS IN URL =====
# Query string: ?page=2&limit=50
params = {"page": 2, "limit": 50, "filter": "active"}
response = requests.get(
"https://api.example.com/resources",
params=params, # requests builds the query string
headers=headers
)
# Actual URL: https://api.example.com/resources?page=2&limit=50&filter=active
# ===== HANDLING ERRORS GRACEFULLY =====
def api_call_with_retry(url, method="GET", max_retries=3, backoff=2):
"""
Make API call with automatic retry on failure.
Args:
url: API endpoint
method: HTTP method (GET, POST, etc.)
max_retries: max number of retries
backoff: seconds to wait between retries
Returns:
response object or None if all retries failed
"""
for attempt in range(max_retries):
try:
if method == "GET":
response = requests.get(url, timeout=10)
elif method == "POST":
response = requests.post(url, timeout=10)
else:
return None
# Check for success
if response.status_code < 400: # 2xx or 3xx
return response
# Retry on 5xx (server error)
if 500 <= response.status_code < 600:
print(f"Server error {response.status_code}, retrying...")
if attempt < max_retries - 1:
time.sleep(backoff * (attempt + 1)) # exponential backoff
continue
# Do not retry on 4xx (client error—our request is wrong)
print(f"Client error {response.status_code}")
return response
except requests.exceptions.Timeout:
print(f"Timeout, attempt {attempt + 1}/{max_retries}")
if attempt < max_retries - 1:
time.sleep(backoff)
except requests.exceptions.ConnectionError:
print(f"Connection error, attempt {attempt + 1}/{max_retries}")
if attempt < max_retries - 1:
time.sleep(backoff)
print("All retries failed")
return None
# ===== CHECKING RESPONSE STATUS =====
response = requests.get("https://api.example.com/data")
# Check status explicitly
if response.status_code == 200:
print("Success")
elif response.status_code == 404:
print("Not found")
elif response.status_code >= 500:
print("Server error")
# Use raise_for_status() to throw exception on 4xx/5xx
try:
response = requests.get("https://api.example.com/data")
response.raise_for_status() # raises HTTPError if status is 4xx/5xx
data = response.json()
except requests.exceptions.HTTPError as e:
print(f"HTTP error: {e}")
except requests.exceptions.RequestException as e:
print(f"Request failed: {e}")
# ===== SESSIONS (REUSE CONNECTION) =====
# For multiple requests, use a session to reuse connection
session = requests.Session()
session.headers.update({"Authorization": f"Bearer {api_token}"})
# All requests through this session use the auth header
response1 = session.get("https://api.example.com/users")
response2 = session.get("https://api.example.com/resources")
response3 = session.post("https://api.example.com/logs", json={"event": "deployment"})
session.close()
# ===== REAL-WORLD EXAMPLE: KUBERNETES RESOURCE QUERY =====
def list_kubernetes_pods(namespace="default"):
"""List all pods in a namespace using Kubernetes API."""
api_token = os.getenv("K8S_TOKEN")
base_url = os.getenv("K8S_API_URL", "https://kubernetes.default.svc.cluster.local")
headers = {
"Authorization": f"Bearer {api_token}",
"Content-Type": "application/json"
}
url = f"{base_url}/api/v1/namespaces/{namespace}/pods"
try:
response = requests.get(url, headers=headers, verify=False, timeout=10)
response.raise_for_status()
pods_data = response.json()
pods = []
for item in pods_data.get("items", []):
pods.append({
"name": item["metadata"]["name"],
"status": item["status"]["phase"],
"ready": any(c["ready"] for c in item["status"]["containerStatuses"])
})
return pods
except requests.exceptions.RequestException as e:
print(f"Failed to list pods: {e}")
return []
# ===== UPLOAD FILE =====
files = {"file": open("report.txt", "rb")}
response = requests.post("https://api.example.com/upload", files=files)
A deployment script queries the Kubernetes API to get current pod status, checks an external monitoring API for health metrics, posts to Slack if health is critical, then patches the Deployment resource to scale replicas and trigger a rolling update. All via HTTP API calls with proper error handling and retries.
Use the JSONPlaceholder free API (https://jsonplaceholder.typicode.com). Write a script that: (1) GETs all users, (2) for each user, GETs their posts, (3) counts posts per user, (4) prints a summary. Handle any network errors gracefully.
Problem: API request works in Postman but fails in your script with "SSL: CERTIFICATE_VERIFY_FAILED".
GET is for reading data (no side effects, can be cached). POST is for creating data or side effects (sending form data, creating resources). GET parameters go in URL; POST data goes in body. GET is idempotent (safe to repeat); POST is not.
response.json() parses the response body as JSON and returns a Python dict (or list). It is equivalent to json.loads(response.text). Throws JSONDecodeError if response is not valid JSON.
Never hardcode it in source code. Read from environment variables: api_key = os.getenv("API_KEY"). Pass in headers: headers={"Authorization": f"Bearer {api_key}"}. Store credentials in secure vaults (AWS Secrets Manager, Azure Key Vault, HashiCorp Vault).
Add a timeout parameter (timeout=10), wrap in try/except for RequestException, implement retry logic with exponential backoff, log each attempt. Distinguish between retryable errors (timeout, 5xx) and non-retryable (4xx). After max retries, fail gracefully or use a fallback.
CI/CD systems query Git APIs. Monitoring sends HTTP requests to health endpoints. Helm uses HTTP to fetch charts. Terraform queries cloud APIs over HTTP. Kubernetes uses HTTP for all cluster interaction. DevOps is 50% orchestrating via APIs.
requests library simplifies HTTP operations: requests.get/post/put/delete. Always check response.status_code and handle errors. Use headers for authentication (Bearer tokens). Parse JSON responses with response.json(). Implement retries with exponential backoff for flaky APIs. Never hardcode secrets—use environment variables or secure vaults. Use sessions to reuse connections across multiple requests. These patterns are essential for cloud automation and API integration.