Talking to APIs, Containers, and the Cloud
Modern infrastructure is managed through APIs, not manual clicks — cloud providers, monitoring tools, container platforms, and CI/CD systems all expose REST APIs that Python can call directly. This module covers the requests library for HTTP calls, the conventions behind cloud SDKs like AWS’s boto3, basic container interaction, and the core idea behind Infrastructure as Code — building directly on the automation skills from M8.
The defining shift in modern DevOps work is treating infrastructure as something you script and version-control, not something you click through in a web console. Every topic in this module supports that shift — calling APIs, automating cloud resources, and codifying infrastructure definitions.
HTTP and REST APIs — The requests Library
A REST API exposes operations over HTTP using a small set of verbs: GET to retrieve data, POST to create something, PUT/PATCH to update, DELETE to remove. Python’s requests library (the de facto standard, though not built-in — pip install requests) makes calling these as simple as a function call.
Host: monitoring.internal
Authorization: Bearer abc123
{“name”: “web-01”,
“cpu”: 45, “status”: “up”}
Every response carries a status code telling you whether the request succeeded and, if not, roughly why. Checking this code is not optional — assuming every API call succeeds is how scripts silently process garbage data or crash three steps downstream from the actual failure.
Authentication — API Keys and Bearer Tokens
Almost every real API requires authentication. The two most common patterns: an API key sent as a header or query parameter, or a Bearer token sent in the Authorization header. Either way, credentials must never be hard-coded directly in your script — exactly the same principle from M5’s config files and M8’s automation scripts: read secrets from environment variables or a secrets manager, never from a literal string in source code.
Retries and Backoff — Handling Unreliable Networks
Networks fail. APIs time out, rate-limit you, or have brief outages. A production script calling external APIs needs a retry strategy — but retrying immediately and repeatedly can make things worse (hammering an already-struggling service). The standard pattern is exponential backoff: wait a little longer after each failed attempt.
fails
fails
✓ succeeds
This directly extends the retry pattern from M6’s error handling — the difference here is the wait time grows (often doubling) with each attempt, giving a struggling service room to recover rather than being retried at a constant, possibly overwhelming rate.
Cloud SDK Conventions — The boto3 Pattern
Cloud providers offer Python SDKs that wrap their APIs into convenient method calls — AWS’s boto3, Azure’s azure-sdk, Google Cloud’s google-cloud libraries. They all follow a similar shape: create a client for a specific service, then call methods on it that map to API operations, returning Python dictionaries.
The pattern is consistent enough that learning one cloud SDK makes the others feel familiar: client = boto3.client(“ec2”), then client.describe_instances() — a method call that, behind the scenes, makes an authenticated HTTP request to AWS and parses the response into a Python dict for you. You rarely touch requests directly when a dedicated SDK exists for the service you’re working with.
Infrastructure as Code — The Underlying Idea
Infrastructure as Code (IaC) means defining servers, networks, and cloud resources in version-controlled files instead of manual console clicks — the same discipline you already apply to application code, applied to infrastructure itself. Tools like Terraform and AWS CloudFormation use declarative configuration files; Python often plays a supporting role, calling cloud SDKs directly for tasks IaC tools don’t cover, or generating/validating IaC configuration programmatically.
requests — GET, POST, headers, and status checking
import requests
import os
# Basic GET request
response = requests.get("https://api.github.com/users/python", timeout=10)
response.status_code # 200
response.json() # parses JSON response body into a Python dict
response.text # raw response body as a string
response.headers # response headers as a dict-like object
# Always check for success before trusting the response
if response.status_code == 200:
data = response.json()
else:
print(f"Request failed: {response.status_code}")
# raise_for_status() — raises an exception automatically on 4xx/5xx
try:
response = requests.get("https://api.example.com/data", timeout=10)
response.raise_for_status()
data = response.json()
except requests.exceptions.HTTPError as e:
print(f"HTTP error: {e}")
except requests.exceptions.Timeout:
print("Request timed out")
except requests.exceptions.ConnectionError:
print("Could not connect — network or DNS issue")
# Authenticated request with a Bearer token (read from environment, never hard-coded)
api_token = os.environ.get("API_TOKEN")
headers = {"Authorization": f"Bearer {api_token}"}
response = requests.get("https://api.example.com/servers", headers=headers, timeout=10)
# POST request with a JSON body
payload = {"name": "new-server", "environment": "staging"}
response = requests.post("https://api.example.com/servers", json=payload, headers=headers, timeout=10)
# json=payload automatically serialises the dict and sets Content-Type: application/json
# Query parameters (the part after ? in a URL)
params = {"environment": "production", "limit": 10}
response = requests.get("https://api.example.com/servers", params=params, timeout=10)
# sends: https://api.example.com/servers?environment=production&limit=10Retry with exponential backoff
import requests
import time
def get_with_retry(url, max_attempts=4, base_delay=1):
"""
GET a URL with exponential backoff on failure.
Wait times: 1s, 2s, 4s, 8s... doubling each attempt.
"""
for attempt in range(1, max_attempts + 1):
try:
response = requests.get(url, timeout=10)
response.raise_for_status()
return response.json()
except (requests.exceptions.RequestException) as e:
if attempt == max_attempts:
raise # out of retries — let the caller handle it
wait_time = base_delay * (2 ** (attempt - 1)) # 1, 2, 4, 8...
print(f"Attempt {attempt} failed ({e}), retrying in {wait_time}s...")
time.sleep(wait_time)
# Usage
data = get_with_retry("https://api.example.com/health")Cloud SDK pattern (boto3 example — AWS)
import boto3
# Create a client for a specific AWS service
ec2 = boto3.client("ec2", region_name="ap-south-1")
# Method calls map to API operations, returning Python dicts
response = ec2.describe_instances()
for reservation in response["Reservations"]:
for instance in reservation["Instances"]:
print(f"{instance['InstanceId']}: {instance['State']['Name']}")
# Starting/stopping resources — same client-and-method pattern everywhere
ec2.start_instances(InstanceIds=["i-0abc123def456"])
ec2.stop_instances(InstanceIds=["i-0abc123def456"])
# S3 (object storage) — same client pattern, different service
s3 = boto3.client("s3")
s3.upload_file("backup.zip", "my-backup-bucket", "backups/backup.zip")
objects = s3.list_objects_v2(Bucket="my-backup-bucket")
for obj in objects.get("Contents", []):
print(f"{obj['Key']}: {obj['Size']} bytes")
# Credentials are read automatically from environment variables,
# ~/.aws/credentials, or an IAM role — never hard-code them in the scriptBasic container interaction (Docker SDK)
import docker
client = docker.from_env() # connects to the local Docker daemon
# List running containers
for container in client.containers.list():
print(f"{container.name}: {container.status}")
# Start a new container
container = client.containers.run(
"nginx:latest", detach=True, name="web-test", ports={"80/tcp": 8080}
)
# Inspect, get logs, stop, remove
container.reload() # refresh container's status info
logs = container.logs(tail=50).decode("utf-8")
container.stop()
container.remove()These examples favour patterns over exact provider details — the same shape applies whether you’re calling AWS, Azure, GCP, or an internal company API.
Checks the health of several internal API endpoints, correctly distinguishing between “service is down” (connection error), “service is slow” (timeout), and “service responded but reports unhealthy” (a 200 with an unhealthy status in the body) — three meaningfully different failure modes that a naive script would lump together.
import requests
from datetime import datetime
def check_endpoint(name, url, timeout=5):
"""
Check one health endpoint.
Returns a dict describing the outcome — never raises.
"""
try:
response = requests.get(url, timeout=timeout)
except requests.exceptions.Timeout:
return {"name": name, "status": "TIMEOUT", "detail": f"No response within {timeout}s"}
except requests.exceptions.ConnectionError:
return {"name": name, "status": "DOWN", "detail": "Connection refused or DNS failure"}
if response.status_code != 200:
return {"name": name, "status": "ERROR", "detail": f"HTTP {response.status_code}"}
try:
body = response.json()
except ValueError:
return {"name": name, "status": "ERROR", "detail": "Response was not valid JSON"}
if body.get("status") != "healthy":
return {"name": name, "status": "UNHEALTHY", "detail": body.get("status", "unknown")}
return {"name": name, "status": "OK", "detail": "healthy"}
# ── Run against several endpoints ──
endpoints = [
("auth-service", "https://auth.internal/health"),
("payments-service", "https://payments.internal/health"),
("search-service", "https://search.internal/health"),
]
print(f"Health check — {datetime.now().strftime('%H:%M:%S')}\n")
for name, url in endpoints:
result = check_endpoint(name, url)
icon = "✓" if result["status"] == "OK" else "✗"
print(f" {icon} {result['name']:20} {result['status']:10} {result['detail']}")✓ auth-service OK healthy
✗ payments-service TIMEOUT No response within 5s
✗ search-service UNHEALTHY degraded
Many APIs return data in pages rather than all at once. This function follows pagination automatically until all results are collected — a pattern needed constantly when pulling complete data sets from any real-world API.
import requests
def fetch_all_pages(base_url, headers=None, page_size=50):
"""
Fetch every page of results from a paginated API.
Assumes the API responds with: {"results": [...], "next_page": int or None}
"""
all_results = []
page = 1
while True:
response = requests.get(
base_url,
params={"page": page, "page_size": page_size},
headers=headers,
timeout=10
)
response.raise_for_status()
body = response.json()
all_results.extend(body["results"])
if body.get("next_page") is None:
break # no more pages — stop
page = body["next_page"]
return all_results
# Simulated usage (against a real paginated endpoint):
# servers = fetch_all_pages("https://api.example.com/servers")
# print(f"Fetched {len(servers)} servers across all pages")
# ── Demonstration with a fake in-memory "API" to show the logic ──
def simulate_paginated_response(page, page_size):
all_servers = [f"server-{i:03d}" for i in range(1, 127)] # 126 total servers
start = (page - 1) * page_size
chunk = all_servers[start:start + page_size]
next_page = page + 1 if start + page_size < len(all_servers) else None
return {"results": chunk, "next_page": next_page}
collected, page = [], 1
while True:
body = simulate_paginated_response(page, page_size=50)
collected.extend(body["results"])
print(f"Page {page}: got {len(body['results'])} items (total so far: {len(collected)})")
if body["next_page"] is None:
break
page = body["next_page"]
print(f"\nTotal servers fetched: {len(collected)}")Page 2: got 50 items (total so far: 100)
Page 3: got 26 items (total so far: 126)
Total servers fetched: 126
Scans cloud compute instances for cost-control issues — specifically, instances running outside business hours that should probably be stopped. Written against the boto3 client pattern; the logic transfers directly to Azure or GCP SDKs with different method names.
import boto3
from datetime import datetime
def find_untagged_instances(region="ap-south-1"):
"""
Return running EC2 instances that are missing a required 'Owner' tag —
a common cost-governance check in real cloud environments.
"""
ec2 = boto3.client("ec2", region_name=region)
response = ec2.describe_instances(
Filters=[{"Name": "instance-state-name", "Values": ["running"]}]
)
untagged = []
for reservation in response["Reservations"]:
for instance in reservation["Instances"]:
tags = {t["Key"]: t["Value"] for t in instance.get("Tags", [])}
if "Owner" not in tags:
untagged.append({
"id": instance["InstanceId"],
"type": instance["InstanceType"],
"launch_time": instance["LaunchTime"],
})
return untagged
# ── In a real AWS account ──
# untagged = find_untagged_instances()
# for inst in untagged:
# print(f"⚠ {inst['id']} ({inst['type']}) — missing Owner tag, launched {inst['launch_time']}")
# ── Demonstration with simulated response data (same shape as the real API) ──
simulated_response = {
"Reservations": [
{"Instances": [
{"InstanceId": "i-0a1b2c3d", "InstanceType": "t3.large",
"LaunchTime": datetime(2026, 5, 2), "Tags": [{"Key": "Owner", "Value": "team-platform"}]},
{"InstanceId": "i-0e5f6g7h", "InstanceType": "m5.xlarge",
"LaunchTime": datetime(2026, 3, 14), "Tags": []},
]}
]
}
untagged = []
for reservation in simulated_response["Reservations"]:
for instance in reservation["Instances"]:
tags = {t["Key"]: t["Value"] for t in instance.get("Tags", [])}
if "Owner" not in tags:
untagged.append(instance)
print(f"Found {len(untagged)} untagged running instance(s):")
for inst in untagged:
print(f" ⚠ {inst['InstanceId']} ({inst['InstanceType']}) — launched {inst['LaunchTime'].date()}")⚠ i-0e5f6g7h (m5.xlarge) — launched 2026-03-14
A complete, reusable API client function with timeout protection, exponential backoff, and proper distinction between retryable errors (timeouts, 5xx, 429) and non-retryable ones (4xx client errors like 404, which retrying won’t fix).
import requests
import time
class APIError(Exception):
"""Raised when an API call fails permanently after all retries."""
pass
RETRYABLE_STATUS_CODES = {429, 500, 502, 503, 504}
def call_api(url, max_attempts=4, timeout=8):
"""
Call an API with retries on transient failures only.
Client errors (4xx other than 429) fail immediately — retrying won't help.
"""
for attempt in range(1, max_attempts + 1):
try:
response = requests.get(url, timeout=timeout)
except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e:
if attempt == max_attempts:
raise APIError(f"Failed after {max_attempts} attempts (network): {e}") from e
wait = 2 ** (attempt - 1)
print(f" Attempt {attempt}: network error, retrying in {wait}s...")
time.sleep(wait)
continue
if response.status_code == 200:
return response.json()
if response.status_code in RETRYABLE_STATUS_CODES:
if attempt == max_attempts:
raise APIError(f"Failed after {max_attempts} attempts: HTTP {response.status_code}")
wait = 2 ** (attempt - 1)
print(f" Attempt {attempt}: HTTP {response.status_code} (retryable), retrying in {wait}s...")
time.sleep(wait)
else:
# Non-retryable client error (e.g. 404) — fail immediately, no point retrying
raise APIError(f"Non-retryable error: HTTP {response.status_code}")
# Usage:
# try:
# data = call_api("https://api.example.com/servers/web-01")
# except APIError as e:
# print(f"Could not retrieve server info: {e}")
print("Function defined — ready to call against a real endpoint.")A Python script that reads a Terraform plan exported as JSON and flags risky changes — specifically, any resource deletion — before a human approves the apply step. This is the kind of “glue” script described in Topic 5: Python supporting an IaC tool rather than replacing it.
import json
def analyse_terraform_plan(plan_json):
"""
Analyse a Terraform plan (as parsed JSON) and flag risky changes.
Returns a dict with counts and a list of resources being destroyed.
"""
summary = {"create": 0, "update": 0, "delete": 0, "no-op": 0}
deletions = []
for change in plan_json.get("resource_changes", []):
actions = change["change"]["actions"]
if "delete" in actions:
summary["delete"] += 1
deletions.append(change["address"])
elif "create" in actions:
summary["create"] += 1
elif "update" in actions:
summary["update"] += 1
else:
summary["no-op"] += 1
return {"summary": summary, "deletions": deletions}
# ── Simulated Terraform plan output (terraform show -json plan.tfplan) ──
sample_plan = {
"resource_changes": [
{"address": "aws_instance.web_01", "change": {"actions": ["create"]}},
{"address": "aws_instance.db_01", "change": {"actions": ["update"]}},
{"address": "aws_security_group.legacy", "change": {"actions": ["delete"]}},
{"address": "aws_s3_bucket.archive_2023", "change": {"actions": ["delete"]}},
{"address": "aws_instance.cache_01", "change": {"actions": ["no-op"]}},
]
}
result = analyse_terraform_plan(sample_plan)
print("Plan summary:")
for action, count in result["summary"].items():
print(f" {action:8}: {count}")
if result["deletions"]:
print(f"\n⚠ WARNING: {len(result['deletions'])} resource(s) will be DESTROYED:")
for addr in result["deletions"]:
print(f" - {addr}")
print("\nManual review required before applying this plan.")create : 1
update : 1
delete : 2
no-op : 1
⚠ WARNING: 2 resource(s) will be DESTROYED:
– aws_security_group.legacy
– aws_s3_bucket.archive_2023
Manual review required before applying this plan.
Exercises 1 and 2 use real public APIs that need no authentication, so you can run them immediately. pip install requests first if you haven’t already.
📋 API Health Dashboard CLI
Build a script called health_dashboard.py that checks multiple API endpoints (use a mix of real public APIs — GitHub’s API, a JSON placeholder API, etc. — and at least one deliberately broken URL) and produces a structured health report, combining argparse from M8 with the API patterns from this module.
- Configuration — read a list of endpoints to check from a JSON config file: [{“name”: “github”, “url”: “https://api.github.com”}, …] with at least 4 entries, including one bad URL that will genuinely fail.
- Argument parsing — –config (path to the JSON file, required), –timeout (default 5), –retries (default 3), –output (path for the JSON report, default health_report.json).
- check_with_retry(name, url, timeout, max_attempts) — checks one endpoint with exponential backoff on timeouts/connection errors, but fails immediately (no retry) on a 404 — echoing Example 4’s distinction between retryable and non-retryable failures. Returns a result dict with name, status (“OK”/”DOWN”/”ERROR”), response_time_ms, and attempts_used.
- Run all checks — check every configured endpoint, printing progress to the terminal as each completes.
- Write the JSON report — to the –output path: a timestamp, overall summary (total/healthy/down counts), and the full per-endpoint results list.
- Exit code — echoing M8’s discipline: exit 0 if every endpoint is healthy, exit 1 if any endpoint is down or erroring, so this script could be wired into a monitoring system’s alert pipeline.
Several of these test the judgment behind retry logic — knowing when retrying helps and when it’s pointless or harmful is more important than the syntax itself.