It's 2 AM. A payment gateway your company depends on starts timing out on and off. You could SSH into ten servers and grep through logs by hand, or you could already have a Python script that checks every dependency, calculates the current error rate, and tells you exactly which service broke - before you've even finished your coffee. That is the real difference between someone who operates a system by clicking around, and someone who builds tools that operate the system for them. Bash is great for a quick one-liner. But the moment your task needs retries, error handling, structured output, or has to run safely every night without a human watching it, Bash starts falling apart and Python takes over. ### What "Production Python" Actually Means Writing a script that works once on your laptop is easy. Writing a script that keeps working when the network hiccups, when a server returns garbage instead of JSON, when someone runs it twice at the same time, or when it has to run unattended at 3 AM every single night - that is a different skill entirely. That skill is what this module is about. Everything here is built around five real jobs an SRE does constantly: * Talk to APIs and services reliably, even when they are flaky * Automate repetitive operational tasks - checks, cleanups, backups, reports * Build small command-line tools that other engineers on the team can actually use * Write code that fails safely and tells you exactly what went wrong * Package and ship that code so it keeps working next month, not just today ### How This Fits Into the Bigger Picture Every module later in this roadmap leans on what you learn here. The error budget calculator you build for SLOs, the chaos engineering scripts, the automation tooling for toil reduction - all of it is Python doing exactly what you are about to learn: calling an API, handling failure gracefully, and producing a clean report a human or another system can read. > 📌 **Remember:** You do not need to become a professional software engineer to be a great SRE. You need to write code that is boring, predictable, and honest about failure. That is the entire goal of this module. ---
Before automating anything, you need the small set of Python building blocks that show up in almost every SRE script ever written. This is not a full "learn Python from zero" course - it is the fast, practical version aimed at what you will actually use. ### Variables, Data Types, and Why They Matter Here A variable is just a labeled box that holds a value. In operational scripts, you constantly work with a handful of types: ```python ## A single value - a string (text) server_name = "prod-api-01" ## A whole number - used for counts, ports, retry attempts retry_count = 3 ## A decimal number - used for percentages, latencies, thresholds error_rate = 0.023 # this means 2.3% ## True or False - used for flags and conditions is_healthy = True ## A list - an ordered collection you can loop over servers = ["prod-api-01", "prod-api-02", "prod-api-03"] ## A dictionary - key-value pairs, like a labeled filing cabinet server_status = { "prod-api-01": "healthy", "prod-api-02": "degraded", } ``` > **Note:** A dictionary (`dict`) is the single most useful data type in operational Python. Almost every API response you will ever parse comes back as a dictionary or a list of dictionaries. Get comfortable reading and writing them. ### Functions - Packaging Logic You Will Reuse A function is a named, reusable block of code. Instead of copy-pasting the same health check logic five times across five scripts, you write it once as a function and call it wherever you need it. ```python def check_server_health(response_code): """ Decide if a server is healthy based on its HTTP response code. Returns True if healthy, False otherwise. """ # Anything in the 200-299 range counts as a successful response return 200 <= response_code < 300 ``` ```python ## Using the function if check_server_health(200): print("Server is healthy") else: print("Server needs attention") ``` The docstring (the text in triple quotes right after the function definition) is not optional decoration - it is what lets another engineer, or you in six months, understand what the function does without reading every line inside it. ### Error Handling - The Skill That Separates Scripts From Tools A script that crashes the instant something unexpected happens is not production-ready. Error handling means telling Python exactly what to do when something goes wrong, instead of letting the whole program die. ```python def get_disk_usage_percent(path): """ Return the disk usage percentage for a given path. Returns None if the path cannot be checked, instead of crashing. """ import shutil try: total, used, free = shutil.disk_usage(path) return round((used / total) * 100, 2) except FileNotFoundError: # The path does not exist - this is expected sometimes, not a crash print(f"Warning: path {path} does not exist") return None except PermissionError: # We are not allowed to read this path print(f"Warning: no permission to check {path}") return None ``` Without try/except: Function hits an error -> entire script crashes -> nothing else runs With try/except: Function hits an error -> catches it -> logs a clear warning -> script keeps going > 🔴 **Common Mistake:** Writing `except Exception: pass` to silently swallow every possible error. This hides real problems instead of handling them - if disk usage checking is genuinely broken, you want to know, not silently get a wrong report. Always catch the specific error you expect, and always at least log what happened. ### Modules - Not Reinventing the Wheel Python ships with a huge standard library, and the operational world has added excellent third-party packages on top of it. You rarely write things from scratch. ```python import os # interact with the operating system - files, paths, environment variables import sys # command-line arguments, exit codes import json # read and write JSON data - the format almost every API uses import time # sleep, timestamps, measuring how long something takes import subprocess # run shell commands from inside Python import logging # proper structured logging instead of print statements ``` ```python ## Reading an environment variable safely - common for API keys and config api_key = os.environ.get("API_KEY", "") # returns "" if the variable is not set if not api_key: print("Error: API_KEY environment variable is not set") sys.exit(1) # exit code 1 tells anything watching this script that it failed ``` > 💡 **Tip:** Exit codes matter more than most beginners realize. `sys.exit(0)` means success, any non-zero number means failure. Monitoring systems, cron jobs, and CI/CD pipelines all read this exit code to decide whether your script worked. A script that always exits with 0 - even when it failed - is a script that will let real problems slip through unnoticed. ### Logging - Why print() Stops Being Enough Every example so far has used `print()` to show output, because it is the simplest way to see what a script is doing while you learn. But `print()` has no concept of severity, no timestamp, and no way to send output somewhere other than the terminal in front of you. The moment a script runs unattended - in a cron job, in a container, on a server you are not watching - `print()` output either disappears or becomes an unstructured wall of text nobody can search. print("Checking server") | Fine for a human watching the terminal right now | Useless three days later when you need to know exactly when this happened, how severe it was, and be able to search across a thousand other lines like it `logging` solves this with almost no extra effort: ```python import logging ## Configure once, at the top of your script logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s" ) logger = logging.getLogger(__name__) logger.info("Checking server prod-api-01") logger.warning("Retrying request after timeout") logger.error("API request failed after all retries") ``` ```text ## What that actually prints 2026-02-14 09:12:03 INFO Checking server prod-api-01 2026-02-14 09:12:08 WARNING Retrying request after timeout 2026-02-14 09:12:15 ERROR API request failed after all retries ``` Each log line now carries a timestamp and a severity level automatically. This matters because it is the exact same information a centralized logging system (the kind covered in the Observability module) expects to receive and index. print() -> fine for quick debugging on your own laptop logging -> what every script that runs unattended in production should use | INFO -> normal operation, useful context WARNING -> something unexpected happened, but the script recovered ERROR -> something failed and needs attention > 💡 **Tip:** You do not need a complicated logging setup to get the benefit. Swapping every `print()` in a script for `logger.info()` or `logger.error()` (matching the right severity) is a five-minute change that makes the script dramatically easier to operate. ### Running System Commands Safely Sometimes Python needs to run an actual shell command - checking `kubectl get pods`, running `df -h`, calling a tool that has no Python library. The `subprocess` module does this, but it has one setting that can turn a simple script into a security hole. ```python import subprocess ## The safe way - command and arguments passed as a list result = subprocess.run( ["kubectl", "get", "pods", "-n", "production"], capture_output=True, text=True, check=True, # raises an error automatically if the command fails timeout=10 # never let a hung command block forever ) print(result.stdout) ``` The dangerous version looks almost identical but behaves completely differently: ```python ## DANGEROUS - never do this with any value that came from user input namespace = input("Enter namespace: ") subprocess.run(f"kubectl get pods -n {namespace}", shell=True) ``` With `shell=True` and a string built from user input, someone could type a namespace like `prod; rm -rf /` and that entire string gets handed to the shell to interpret - including the part after the semicolon. This is called command injection, and it is one of the most common real-world security bugs in operational scripts. subprocess.run(["cmd", "arg1", "arg2"]) -> safe, each piece is a separate argument subprocess.run("cmd arg1 arg2", shell=True) -> risky, the shell interprets the whole string > ⚠️ **Security:** Pass commands as a list of arguments, not as one combined string with `shell=True`. If you must build a command from a variable, that variable should go into the list as its own element, never concatenated into a shell string. ### Structured Output - Making Your Script's Results Machine-Readable A script printing human-friendly sentences is fine when a person is reading it directly. But SRE automation is usually read by *other programs* - a monitoring system, a CI/CD pipeline, a log aggregator. Those tools need predictable, structured data, not sentences designed for a human. ```python import json def build_health_report(service_name, status, error_rate, latency_ms): """Build a structured report any downstream tool can parse reliably.""" return { "service": service_name, "status": status, "error_rate": error_rate, "latency_ms": latency_ms, } report = build_health_report("payment-api", "degraded", 2.31, 842) print(json.dumps(report)) ``` ```text ## What gets printed - one clean JSON line {"service": "payment-api", "status": "degraded", "error_rate": 2.31, "latency_ms": 842} ``` Compare that to printing `"payment-api is degraded with 2.31% errors and 842ms latency"` - a human can read that sentence, but a monitoring system trying to extract the error rate from it would need fragile text parsing that breaks the moment the wording changes slightly. The JSON version never breaks that way, because every field has an explicit name. > 📌 **Remember:** When a script's output will be read by another program - a pipeline, a monitoring tool, a log shipper - print JSON, not sentences. When the output is only ever read by a human sitting at the terminal, plain text is fine. Know which audience you are writing for. ---
Almost every SRE task eventually needs to talk to something over HTTP - a monitoring system, a cloud provider, an internal service, a chat app to send an alert. The `requests` library is the standard tool for this in Python. ### Making Your First API Call ```python import requests ## A simple GET request - asking a server for information response = requests.get("https://api.github.com/repos/python/cpython") ## Check if the request actually succeeded before trusting the data if response.status_code == 200: data = response.json() # convert the JSON response into a Python dictionary print(f"Stars: {data['stargazers_count']}") else: print(f"Request failed with status code {response.status_code}") ``` > **Note:** `response.status_code` is the HTTP status code - 200 means success, 404 means not found, 500 means the server itself is broken. Never assume a request worked. Always check the status code before trusting the response body. ### Authentication - Proving Who You Are Most real APIs require you to prove your identity before they give you anything useful. The two patterns you will see constantly: ```python ## Pattern 1 - API key in a header (very common) headers = {"Authorization": f"Bearer {api_key}"} response = requests.get("https://api.example.com/servers", headers=headers) ## Pattern 2 - basic username and password authentication response = requests.get( "https://api.example.com/servers", auth=("username", "password") ) ``` > ⚠️ **Security:** Never hardcode an API key or password directly into your script. If that script ever gets committed to Git, shared with a teammate, or pasted into a chat, that credential is now compromised. Always load secrets from environment variables or a secrets manager, never from a string typed directly into your code. ### Pagination - When One Request Is Not Enough Real APIs almost never return everything in one response. A list of 10,000 servers might come back 100 at a time, and it is your job to keep asking for the next page until there is nothing left. ```python def get_all_servers(base_url, headers): """ Fetch every server from a paginated API, one page at a time, and return the combined list. """ all_servers = [] next_page_url = f"{base_url}/servers?page=1" while next_page_url: response = requests.get(next_page_url, headers=headers) data = response.json() all_servers.extend(data["results"]) # add this page's results to our list next_page_url = data.get("next_page_url") # None once there are no more pages return all_servers ``` Page 1 -> 100 servers -> "next_page_url" points to page 2 Page 2 -> 100 servers -> "next_page_url" points to page 3 Page 3 -> 47 servers -> "next_page_url" is None -> loop stops Total collected: 247 servers ### Rate Limiting - Respecting the API's Limits APIs limit how many requests you can make per minute to protect themselves from being overwhelmed. If you blow past that limit, the API starts rejecting your requests - usually with an HTTP 429 status code ("Too Many Requests"). ```python import time def call_api_with_rate_limit_respect(url, headers, max_attempts=3): """ Make an API call and automatically wait if we hit a rate limit. Capped at max_attempts so a persistently rate-limited endpoint cannot turn into an infinite loop. """ for attempt in range(1, max_attempts + 1): response = requests.get(url, headers=headers) if response.status_code != 429: return response # The API tells us how long to wait using the Retry-After header wait_seconds = int(response.headers.get("Retry-After", 5)) print(f"Rate limited (attempt {attempt}/{max_attempts}). Waiting {wait_seconds}s.") time.sleep(wait_seconds) # Ran out of attempts while still rate-limited - stop and report it return response ``` > 📌 **Remember:** A 429 status code is not really a failure - it is the API asking you politely to slow down. Handling it gracefully by waiting and retrying is standard, expected behavior. But always cap the number of attempts. A retry loop with no ceiling is not resilience, it is a script that can hang forever if the endpoint stays rate-limited. ### Webhooks - When the API Calls You Instead So far you have been the one reaching out to an API. A webhook flips this around - an external service (like GitHub, a monitoring tool, or a payment processor) sends *you* a message the moment something happens, instead of you having to constantly ask "did anything change yet?" ```python from flask import Flask, request, jsonify app = Flask(__name__) @app.route("/webhook/alert", methods=["POST"]) def receive_alert(): """ Receives an alert payload pushed to us by a monitoring tool. """ payload = request.get_json() alert_name = payload.get("alert_name", "unknown") severity = payload.get("severity", "unknown") print(f"Received alert: {alert_name} (severity: {severity})") # Always respond quickly - the sender is often waiting for confirmation return jsonify({"status": "received"}), 200 ``` > **Note:** A webhook receiver should do the absolute minimum work needed to acknowledge receipt, then hand the real processing off elsewhere (a queue, a background job). If your webhook handler is slow, the sender may assume delivery failed and resend the same alert repeatedly. ---
Networks fail. Servers get overloaded. DNS occasionally has a bad moment. None of this means your entire script should crash. This is the single most important operational skill in this whole module - code that survives a flaky world instead of falling over the first time something hiccups. ### Timeouts - Never Wait Forever By default, a network request in Python will wait indefinitely for a response if you do not tell it otherwise. A single hung server can freeze your entire script forever. ```python try: ## Always set a timeout - this one waits max 5 seconds before giving up response = requests.get("https://api.example.com/status", timeout=5) except requests.exceptions.Timeout: print("Request timed out - the server took too long to respond") ``` > 🔴 **Common Mistake:** Calling `requests.get(url)` with no timeout at all. If that server never responds, your script hangs forever - not for five minutes, forever. Every single network call you write should have an explicit timeout. ### Retries With Exponential Backoff - The Professional Way to Try Again If a request fails, trying again immediately is often the worst thing you can do - if the server is overloaded, hammering it again instantly just makes things worse. Exponential backoff means you wait a little longer after each failed attempt. Attempt 1 fails -> wait 1 second -> try again Attempt 2 fails -> wait 2 seconds -> try again Attempt 3 fails -> wait 4 seconds -> try again Attempt 4 fails -> wait 8 seconds -> give up, report failure ```python import time import random def call_with_retry(url, max_attempts=4): """ Call an API with exponential backoff retries. Each failed attempt waits longer than the last before trying again. """ for attempt in range(1, max_attempts + 1): try: response = requests.get(url, timeout=5) response.raise_for_status() # raises an error for 4xx/5xx status codes return response except requests.exceptions.RequestException as error: if attempt == max_attempts: # This was the last attempt - give up and let the caller know raise wait_time = (2 ** attempt) + random.uniform(0, 1) # backoff + jitter print(f"Attempt {attempt} failed ({error}). Retrying in {wait_time:.1f}s") time.sleep(wait_time) ``` > **Note:** The small random amount added to the wait time is called "jitter." Imagine 100 servers all failing at the exact same moment and all retrying after exactly 2 seconds - they would all hammer the recovering service at the exact same instant again. Jitter spreads those retries out so they don't all pile on at once. ### Not Every Failure Deserves a Retry The retry code above catches `RequestException` and retries no matter what went wrong. That is a trap. Some failures are worth retrying. Others will fail exactly the same way every single time, and retrying them just wastes time and adds noise to your logs. HTTP status code comes back | ┌───────┴────────┐ | | retryable permanent | | 429 - rate limited 400 - bad request (your data is wrong) 502 - bad gateway 401 - unauthorized (your credentials are wrong) 503 - unavailable 403 - forbidden (you don't have access) 504 - gateway timeout 404 - not found (that thing does not exist) timeout / connection 422 - the server understood but refuses it reset errors A 500-series error or a timeout often means the server (or the network) is having a temporary bad moment - trying again in a few seconds might succeed. A 400-series error almost always means something about *your request* is wrong, and it will still be wrong on the second, third, and fiftieth attempt. Retrying it just delays you from noticing the real bug. ```python ## Status codes worth retrying - temporary, server-side, or network problems RETRYABLE_STATUS_CODES = {429, 500, 502, 503, 504} def call_with_smart_retry(url, max_attempts=4): """ Retries only failures that are actually likely to succeed on a second attempt. Client errors (4xx other than 429) fail immediately instead of wasting time retrying something that will never change. """ for attempt in range(1, max_attempts + 1): try: response = requests.get(url, timeout=5) if response.status_code not in RETRYABLE_STATUS_CODES: # Either success, or a permanent client error - stop here either way response.raise_for_status() return response if attempt == max_attempts: response.raise_for_status() wait_time = (2 ** attempt) + random.uniform(0, 1) print(f"Retryable failure ({response.status_code}). Retrying in {wait_time:.1f}s") time.sleep(wait_time) except requests.exceptions.Timeout: # Timeouts and connection errors are also worth retrying if attempt == max_attempts: raise wait_time = (2 ** attempt) + random.uniform(0, 1) time.sleep(wait_time) ``` > 🔴 **Common Mistake:** Wrapping every network call in a blanket `except requests.exceptions.RequestException: retry`. This treats "your password is wrong" the same as "the server is briefly overloaded" - one of those needs a fix, not a retry loop. ### Circuit Breakers - Stop Hitting a Service That Is Already Down Retrying is good when a failure is temporary. But if a service has been failing for the last 50 requests in a row, retrying again is pointless - you are just wasting time and adding load to something that is clearly broken. A circuit breaker "trips open" after too many failures and stops trying for a while, like a real electrical circuit breaker cutting power to prevent damage. Every real circuit breaker - the kind SREs actually mean when they use the term - moves through three distinct states, not just "working" and "broken": CLOSED (normal operation, calls go through) | | too many failures in a row v OPEN (calls blocked immediately, no network call made at all) | | cooldown period expires v HALF-OPEN (let exactly one test call through) | ├── that test call succeeds -> back to CLOSED | └── that test call fails -> back to OPEN, cooldown restarts The HALF-OPEN state is the piece easy to miss. Without it, a breaker either blocks everything or lets everything through - there is no safe way to *check* if the dependency has recovered without risking a flood of requests hitting it the moment the cooldown ends. HALF-OPEN sends exactly one probe and decides based on that single result. ```python import time class CircuitBreaker: """ A three-state circuit breaker: CLOSED, OPEN, HALF_OPEN. This mirrors the standard pattern SRE and distributed-systems teams mean when they say "circuit breaker." """ CLOSED = "CLOSED" OPEN = "OPEN" HALF_OPEN = "HALF_OPEN" def __init__(self, failure_threshold=5, cooldown_seconds=30): self.failure_threshold = failure_threshold self.cooldown_seconds = cooldown_seconds self.failure_count = 0 self.state = self.CLOSED self.opened_at = None def allow_request(self): """Returns True if a call should be attempted right now.""" if self.state == self.OPEN: # Has the cooldown passed? If so, allow exactly one probe through. if time.time() - self.opened_at >= self.cooldown_seconds: self.state = self.HALF_OPEN return True return False # CLOSED and HALF_OPEN both allow the call through return True def record_success(self): # A success from any state resets everything back to normal self.failure_count = 0 self.state = self.CLOSED def record_failure(self): self.failure_count += 1 if self.state == self.HALF_OPEN: # The probe call failed - the dependency is still broken, reopen immediately self.state = self.OPEN self.opened_at = time.time() return if self.failure_count >= self.failure_threshold: self.state = self.OPEN self.opened_at = time.time() ``` ```python breaker = CircuitBreaker(failure_threshold=5, cooldown_seconds=30) def call_service_safely(url): if not breaker.allow_request(): print(f"Circuit breaker is {breaker.state} - skipping call") return None try: response = requests.get(url, timeout=5) response.raise_for_status() breaker.record_success() return response except requests.exceptions.RequestException: breaker.record_failure() return None ``` CLOSED: calls go through normally, failures are counted 5 failures in a row -> state becomes OPEN -> calls blocked for 30 seconds 30 seconds pass -> state becomes HALF_OPEN -> exactly ONE test call allowed through Test call succeeds -> state becomes CLOSED -> normal operation resumes Test call fails -> state goes back to OPEN -> cooldown restarts from zero ### Idempotency - Making It Safe to Retry Here's the problem retries create: if you retry a request that actually succeeded on the server side but the response got lost on the way back to you, you might accidentally do the same thing twice - like charging a customer's card twice for one purchase. Idempotency means designing an operation so that running it multiple times has the exact same effect as running it once. **Not every operation has the same relationship with retries.** Some HTTP methods are naturally safe to repeat by their own definition. Others are inherently risky unless you take extra steps. Naturally idempotent - safe to retry as-is: GET - just reads data, asking twice changes nothing PUT - replaces a resource with the given data, replacing it twice is the same as once DELETE - removes a resource, deleting something already gone is still "gone" Not naturally idempotent - needs an idempotency key: POST /charge - "charge $50" run twice means charged $100 POST /create-order - run twice means two orders exist POST /send-notification - run twice means the user gets paged twice The dangerous scenario is specifically this one: You send a POST request | Server receives it and processes it successfully | Server's response gets lost on the way back to you (network blip) | Your code sees a timeout - it has NO WAY to know if the server actually did the work | Your retry logic retries the request | Without an idempotency key: the server has no memory of "already did this" -> duplicate operation With an idempotency key: the server recognizes the key -> returns the original result, does nothing new ```python import uuid def charge_customer(customer_id, amount): """ Charge a customer, using an idempotency key so that retrying this exact request never results in a duplicate charge. """ idempotency_key = str(uuid.uuid4()) # generate this ONCE, before any retries headers = {"Idempotency-Key": idempotency_key} payload = {"customer_id": customer_id, "amount": amount} return call_with_retry_and_headers("https://api.payments.com/charge", headers, payload) ``` > 📌 **Remember:** The idempotency key must be generated once, before the first attempt - not regenerated on every retry. If you regenerate it each time, the payment server has no way to know that attempt #2 is a retry of attempt #1 rather than a brand new charge. This is exactly why the key is created outside the retry loop in the code above, not inside it. ---
Untested automation is dangerous automation. A script that deletes old log files, restarts a service, or modifies a database is exactly the kind of code you want to be confident about before it ever runs against production. ### Unit Tests - Checking One Piece at a Time A unit test checks that a single function behaves correctly, in isolation, without needing a real server or database. ```python ## file: health_check.py def is_healthy(status_code): return 200 <= status_code < 300 ``` ```python ## file: test_health_check.py from health_check import is_healthy def test_success_status_is_healthy(): assert is_healthy(200) == True def test_server_error_is_not_healthy(): assert is_healthy(500) == False def test_boundary_299_is_healthy(): assert is_healthy(299) == True def test_boundary_300_is_not_healthy(): assert is_healthy(300) == False ``` ```bash ## Run all tests in the current directory using pytest pytest test_health_check.py -v ``` > **Note:** Notice the boundary tests - checking exactly at 299 and 300. Bugs love to hide at boundaries. Testing only "obviously true" and "obviously false" cases misses the exact spot where logic errors usually live. ### Mocking - Testing Without Calling the Real API You do not want your test suite making a real API call every time it runs - it would be slow, could fail due to network issues unrelated to your code, and might even cost money or affect real systems. Mocking replaces the real API call with a fake, predictable stand-in. ```python from unittest.mock import patch import requests def get_server_status(url): response = requests.get(url, timeout=5) return response.json()["status"] ``` ```python ## Test using a mock instead of a real network call from unittest.mock import patch, MagicMock @patch("requests.get") def test_get_server_status_healthy(mock_get): # Build a fake response object that behaves like a real one fake_response = MagicMock() fake_response.json.return_value = {"status": "healthy"} mock_get.return_value = fake_response result = get_server_status("https://fake-url.com/status") assert result == "healthy" mock_get.assert_called_once() # confirm requests.get was actually called ``` > 💡 **Tip:** Mock the boundary of your system - the network call, the database query, the file write - not your own logic. You want to test that YOUR code makes the right decisions given a certain response, not re-test that `requests` itself works. ### Integration Tests - Checking That the Pieces Work Together While unit tests isolate one function, an integration test checks that several pieces work correctly together - for example, that your retry logic, your API client, and your JSON parsing all cooperate correctly end to end. These are slower and less common, usually run against a real test environment rather than production. ```python def test_retry_logic_gives_up_after_max_attempts(): """ Integration-style test: verify the full retry flow actually stops after the configured number of attempts, using a real (but intentionally broken) URL. """ with pytest.raises(requests.exceptions.RequestException): call_with_retry("http://localhost:9999/does-not-exist", max_attempts=2) ``` ---
A script that only you can run by editing variables at the top of the file is not a tool - it is a personal hack. A real CLI (command-line interface) tool takes input as arguments, has a `--help` flag, and can be handed to any teammate without them ever opening the source code. ### argparse - Python's Built-in CLI Framework ```python ## file: check_disk.py import argparse import sys def main(): parser = argparse.ArgumentParser( description="Check disk usage and alert if it exceeds a threshold." ) parser.add_argument( "--path", default="/", help="Filesystem path to check (default: /)" ) parser.add_argument( "--threshold", type=int, default=90, help="Alert if usage percent exceeds this value (default: 90)" ) args = parser.parse_args() usage_percent = get_disk_usage_percent(args.path) if usage_percent is None: print(f"Could not check {args.path}") sys.exit(2) # exit code 2 = could not run the check at all print(f"{args.path}: {usage_percent}% used") if usage_percent > args.threshold: print(f"ALERT: usage exceeds threshold of {args.threshold}%") sys.exit(1) # exit code 1 = check ran fine, but it failed sys.exit(0) # exit code 0 = everything is fine if __name__ == "__main__": main() ``` ```bash ## Running the tool - this is what makes it a real CLI tool python check_disk.py --path /var/log --threshold 80 ## Anyone can see how to use it without reading the code python check_disk.py --help ``` > 📌 **Remember:** Three distinct exit codes above (0, 1, 2) are not random. Monitoring systems and CI pipelines often want to distinguish "check passed," "check failed," and "check could not even run" as three different outcomes. Design your exit codes deliberately. ### Configuration - Where Should a Setting Actually Live? The disk checker above takes `--threshold` as a command-line argument. But real tools usually need to work across several environments without anyone editing code, and settings can come from more than one place. When the same setting could come from several sources at once, you need a clear, predictable order for which one wins. CLI argument -> highest priority (explicit, one-time override) | Environment variable -> next (set once per environment - dev, staging, prod) | Config file -> next (shared team defaults, checked into a repo) | Hardcoded default -> lowest priority (fallback if nothing else is set) ```python import os def get_threshold(cli_value): """ Resolve the threshold setting using a clear priority order: CLI argument > environment variable > hardcoded default. """ if cli_value is not None: return cli_value env_value = os.environ.get("DISK_THRESHOLD") if env_value is not None: return int(env_value) return 90 # the fallback default if nothing else was provided ``` **Configuration and secrets are not the same thing, even though they often live near each other.** A disk usage threshold is configuration - safe to commit to Git, safe to print in a log. An API key or database password is a secret - it should never appear in a config file that gets committed to version control, and it should never be printed in a log line. > ⚠️ **Security:** A `config.yaml` file with your team's default thresholds and timeouts is fine to check into Git. A `config.yaml` file with an API key sitting next to those thresholds is a credential leak waiting to happen. Keep secrets in environment variables or a secrets manager, and keep everything else in ordinary config. ### Graceful Shutdown - Handling Being Stopped Mid-Task A script running unattended will eventually get interrupted - a deployment restarts the container it's running in, someone hits Ctrl+C, a process manager sends it a stop signal. What happens at that exact moment matters. A script that dies mid-write can leave a half-written file, a locked resource, or a partially processed batch of work behind. ```python import signal import sys shutdown_requested = False def handle_shutdown_signal(signum, frame): """ Called when the OS asks this script to stop. Instead of dying immediately, set a flag and let the current unit of work finish cleanly. """ global shutdown_requested print("Shutdown signal received - finishing current task, then exiting") shutdown_requested = True ## Register the handler for both common stop signals signal.signal(signal.SIGTERM, handle_shutdown_signal) # sent by process managers, Kubernetes, systemd signal.signal(signal.SIGINT, handle_shutdown_signal) # sent by Ctrl+C def process_all_servers(servers): for server in servers: if shutdown_requested: print("Stopping early due to shutdown request - no partial work left behind") break check_one_server(server) sys.exit(0) ``` Signal arrives (SIGTERM) | Flag is set - current unit of work is allowed to finish | Loop checks the flag before starting the NEXT unit of work | No new work starts, script exits cleanly instead of being killed mid-task > 📌 **Remember:** You do not need to handle every signal for every script. But any script that writes files, updates a database, or processes a batch of items one at a time benefits enormously from checking a shutdown flag between items, rather than assuming it will always be allowed to run to completion uninterrupted. ### Composable Tools - Designed to Work With Other Tools A defining trait of good command-line tools on Linux is composability - they can be piped into each other. Your Python tools should follow the same philosophy: read from standard input when useful, write clean output to standard output, and send errors to standard error, not mixed together. ```python import sys def main(): ## Read server names from standard input, one per line ## This lets the tool be used like: cat servers.txt | python check_servers.py for line in sys.stdin: server = line.strip() if not server: continue result = check_one_server(server) print(result) # goes to stdout - can be piped to another tool if __name__ == "__main__": main() ``` ```bash ## This tool can now be chained with other Unix tools naturally cat servers.txt | python check_servers.py | grep "DEGRADED" ``` ---
It's 2 AM. A payment gateway your company depends on starts timing out on and off. You could SSH into ten servers and gr...
Before automating anything, you need the small set of Python building blocks that show up in almost every SRE script eve...
Almost every SRE task eventually needs to talk to something over HTTP - a monitoring system, a cloud provider, an intern...
Networks fail. Servers get overloaded. DNS occasionally has a bad moment. None of this means your entire script should c...
Untested automation is dangerous automation. A script that deletes old log files, restarts a service, or modifies a data...
A script that only you can run by editing variables at the top of the file is not a tool - it is a personal hack. A real...
Imagine your disk checker needs to check 1,000 servers, not one. Checking them one at a time, waiting for each response ...
A script that only runs on your laptop because of some package you installed two years ago and forgot about is not somet...
This lab combines everything in this module into one real tool - a CLI that queries a monitoring endpoint, calculates ho...
Task What to Use Read config or secrets os.environ.get("VARNAME") Log with severity and timestamps logging.getLogger(nam...
Aligns directly with DevOps, Site Reliability (SRE), and Platform Engineering job descriptions.