Python Devops

Python for DevOps — Quick Reference

Reference for using Python the way ops teams use it: shelling out, manipulating files, parsing structured data, building CLI tools, writing tests, talking to clouds and Kubernetes.

When to Use

• Logic that doesn't fit one screen of bash (anything past ~50 lines of control flow reads better in Python)
• Working with real data structures: nested dicts, typed records, dataframes
• Writing testable ops automation where pytest is far more pleasant than bash testing
• HTTP, JSON/YAML/TOML/XML parsing, cloud SDKs, and Kubernetes API interaction
• Cross-platform work (especially Windows compatibility) without rewriting

When NOT to Use

• For plain pipe-A-into-B-into-C tasks — bash is shorter and faster to start up
• Simple CLI glue where subprocess calls would add unnecessary complexity
• Performance-critical loops where compiled languages are more appropriate

Mental model

Treat ops Python like any other Python: virtualenvs, packaging, tests. The content is different (you call out to subprocesses, you parse YAML, you talk to clouds) but the discipline that keeps Python maintainable applies the same.

Examples assume Python 3.10+.

Project layout

A small but proper layout that scales:

myops/
├── pyproject.toml
├── src/myops/
│   ├── __init__.py
│   ├── cli.py
│   └── tasks/
│       ├── __init__.py
│       └── deploy.py
├── tests/
│   ├── __init__.py
│   └── test_deploy.py
└── README.md

pyproject.toml (PEP 621):

[project]
name = "myops"
version = "0.1.0"
requires-python = ">=3.10"
dependencies = [
  "click>=8",
  "pyyaml",
  "requests",
]

[project.scripts]
myops = "myops.cli:main"

[project.optional-dependencies]
dev = ["pytest", "pytest-cov", "mypy", "ruff"]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

Virtualenvs and packaging

python -m venv .venv          # create
source .venv/bin/activate     # *nix
.\.venv\Scripts\Activate.ps1   # Windows PowerShell
deactivate                    # leave

python -m pip install -U pip wheel
python -m pip install -e '.[dev]'    # editable install + dev extras
python -m pip install -r requirements.txt

python -m pip freeze > requirements.txt
python -m pip list --outdated

Modern alternatives:

• uv (fast all-in-one): uv venv, uv pip install, uv run.
• poetry (popular for libraries): poetry add, poetry install, poetry run.
• pipx for installing CLI tools globally without polluting envs.

For binary-distributable tools, build wheels (python -m build) or freeze with pyinstaller / shiv.

Calling out to the shell

subprocess.run is the right answer 95% of the time.

import subprocess

# 1) capture output, raise on non-zero
result = subprocess.run(
    ["git", "rev-parse", "HEAD"],
    capture_output=True, text=True, check=True,
)
sha = result.stdout.strip()

# 2) just want exit code, stream output to terminal
subprocess.run(["pytest", "-q"], check=True)

# 3) feed stdin
subprocess.run(["sort"], input="b\na\nc\n", text=True, capture_output=True)

# 4) timeout
subprocess.run(["slow"], timeout=30)

# 5) environment override
subprocess.run(["printenv", "FOO"], env={**os.environ, "FOO": "bar"})

# 6) cwd
subprocess.run(["make", "test"], cwd="/path/to/proj", check=True)

Avoid shell=True unless you actually need shell features (globs, pipes, redirection). When you do, pass a string and accept that you have to think about quoting.

# DO NOT do this with untrusted input
subprocess.run(f"grep {pattern} *.log | head", shell=True)
# DO this
subprocess.run(["grep", pattern, *glob.glob("*.log")], stdout=open("h.txt","w"))

For long-running processes you want to interact with, use Popen:

proc = subprocess.Popen(
    ["tail", "-F", "/var/log/app.log"],
    stdout=subprocess.PIPE, text=True, bufsize=1,
)
for line in proc.stdout:
    if "ERROR" in line:
        notify(line)

Files and paths

pathlib.Path is the modern way; os.path is legacy.

from pathlib import Path

p = Path.home() / "logs" / "app.log"
p.exists()
p.is_file()
p.suffix              # '.log'
p.stem                # 'app'
p.parent              # Path('/home/ada/logs')

# read / write whole files
text = p.read_text(encoding="utf-8")
p.write_text("hello\n", encoding="utf-8")
data = p.read_bytes()

# create / remove
Path("a/b/c").mkdir(parents=True, exist_ok=True)
p.unlink(missing_ok=True)

# walk a tree
for log in Path("/var/log").rglob("*.log"):
    print(log)

# atomic write
import os, tempfile
def atomic_write(path: Path, content: str) -> None:
    tmp = tempfile.NamedTemporaryFile("w", delete=False,
                                      dir=path.parent, encoding="utf-8")
    try:
        tmp.write(content); tmp.flush(); os.fsync(tmp.fileno())
    finally:
        tmp.close()
    os.replace(tmp.name, path)

Globbing:

list(Path(".").glob("*.py"))            # one level
list(Path(".").rglob("*.py"))           # recursive

Text and regex

import re

# match
m = re.match(r"^(\w+)@(\S+)$", "ada@example.com")
m.group(1), m.group(2)             # 'ada', 'example.com'

# search / findall / finditer
re.search(r"\d{4}", "v2025-Q1")
re.findall(r"\b\w+\b", "two words")
for m in re.finditer(r"[A-Z]\w+", text):
    print(m.start(), m.group())

# substitute
re.sub(r"\s+", " ", text).strip()

# pre-compile for hot loops
PAT = re.compile(r"^ERROR\s+(\d+)\s+(.*)$", re.M)
for m in PAT.finditer(log):
    code, msg = m.groups()

String methods you'll use a lot:

"Hello".lower()
"  x  ".strip()
"a,b,c".split(",")
",".join(["a","b","c"])
f"{name=} {count=}"            # debug-style f-string

Structured data: JSON, YAML, TOML, CSV

import json, yaml, csv, tomllib  # Python 3.11+

# JSON
data = json.loads(text)
text = json.dumps(data, indent=2, sort_keys=True)
data = json.load(open("f.json"))
json.dump(data, open("f.json","w"), indent=2)

# YAML (pip install pyyaml)
data = yaml.safe_load(open("f.yaml"))   # ALWAYS safe_load, never load
yaml.safe_dump(data, open("f.yaml","w"), sort_keys=False)

# TOML (read-only stdlib in 3.11+; write with tomli-w)
with open("pyproject.toml","rb") as f:
    cfg = tomllib.load(f)

# CSV
with open("data.csv", newline="") as f:
    reader = csv.DictReader(f)
    rows = [r for r in reader if r["status"] == "OK"]

with open("out.csv","w", newline="") as f:
    w = csv.DictWriter(f, fieldnames=["id","name","status"])
    w.writeheader()
    w.writerows(rows)

For dataframes (analytics, large files), reach for pandas.

CLI tools

click is the friendlier choice; argparse is the stdlib option.

# click
import click

@click.command()
@click.argument("path", type=click.Path(exists=True, dir_okay=False))
@click.option("--n", default=10, show_default=True, type=int)
@click.option("-v", "--verbose", is_flag=True)
def head(path, n, verbose):
    """Print the first N lines of PATH."""
    if verbose: click.echo(f"reading {path}", err=True)
    with open(path) as f:
        for i, line in enumerate(f):
            if i >= n: break
            click.echo(line.rstrip())

if __name__ == "__main__":
    head()

# argparse
import argparse

def main():
    p = argparse.ArgumentParser(description="show first N lines")
    p.add_argument("path")
    p.add_argument("-n", type=int, default=10)
    p.add_argument("-v", "--verbose", action="store_true")
    args = p.parse_args()
    ...

if __name__ == "__main__":
    main()

For a git-style tool with subcommands, click.Group or argparse.add_subparsers both work; click ends up much shorter for deep trees.

Logging and observability

Use the stdlib logging. Don't print().

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s %(message)s",
)
log = logging.getLogger(__name__)

log.debug("started with %s", args)
log.info("connected to %s", host)
log.warning("retrying after %s", err)
log.error("failed: %s", err, exc_info=True)

Conventional levels: DEBUG (developers), INFO (operational), WARNING (unexpected but recoverable), ERROR (operation failed), CRITICAL (process can't continue).

For structured (JSON) logs in production, structlog is the popular choice. For metrics, prometheus_client. For traces, opentelemetry.

# structlog quickstart
import structlog
log = structlog.get_logger()
log.info("deploy_started", service="web", version="1.4.2")

Testing with pytest

# tests/test_deploy.py
import pytest
from myops.tasks.deploy import compute_target

def test_compute_target_default():
    assert compute_target("web") == "web-prod"

@pytest.mark.parametrize("env,expected", [
    ("prod", "web-prod"),
    ("staging", "web-stg"),
    ("dev", "web-dev"),
])
def test_compute_target_envs(env, expected):
    assert compute_target("web", env=env) == expected

@pytest.fixture
def fake_config(tmp_path):
    cfg = tmp_path / "config.yaml"
    cfg.write_text("env: test\n")
    return cfg

def test_loads_config(fake_config):
    assert load_config(fake_config)["env"] == "test"

Run:

pytest -q                       # quiet
pytest -k deploy                # name filter
pytest --maxfail=1 -x           # stop on first failure
pytest --cov=myops              # with pytest-cov
pytest -n auto                  # parallel with pytest-xdist

Mocking:

from unittest.mock import patch, MagicMock

@patch("myops.tasks.deploy.requests.post")
def test_calls_api(mock_post):
    mock_post.return_value.status_code = 200
    deploy("web", "1.0")
    mock_post.assert_called_once()

pyfakefs (fake filesystem) and responses (HTTP mocking) are worth knowing.

HTTP and APIs

import requests

r = requests.get("https://api.example.com/v1/items", timeout=10,
                 headers={"Authorization": f"Bearer {token}"})
r.raise_for_status()
items = r.json()

r = requests.post("https://api/v1/items", json={"name":"x"}, timeout=10)

Use a Session for repeated calls (connection reuse, default headers, retries):

from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

s = requests.Session()
s.headers.update({"Authorization": f"Bearer {token}"})
retry = Retry(total=5, backoff_factor=0.5,
              status_forcelist=[429,500,502,503,504],
              allowed_methods=["GET","POST"])
s.mount("https://", HTTPAdapter(max_retries=retry))

httpx is a modern alternative with sync + async APIs. For high throughput, async + httpx.AsyncClient or aiohttp.

Config and secrets

• Plain config: YAML / TOML files, parsed at startup.
• Per-environment overrides: env vars with a default in code, e.g. os.environ.get("PORT", "8080").
• Secrets: never check them in. Use the cloud's secret manager (AWS Secrets Manager, GCP Secret Manager, HashiCorp Vault) or .env + python-dotenv for local dev.
• pydantic-settings (or dynaconf) gives you a typed Settings object populated from env vars, files and defaults.

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    api_url: str = "https://api.example.com"
    api_token: str             # required; from API_TOKEN env var
    log_level: str = "INFO"

    class Config:
        env_file = ".env"

settings = Settings()

Docker and Kubernetes interaction

# docker SDK
import docker
client = docker.from_env()
client.containers.run("alpine", "echo hi", remove=True)
for c in client.containers.list():
    print(c.short_id, c.status, c.image.tags)

# Kubernetes client
from kubernetes import client, config
config.load_kube_config()              # ~/.kube/config
v1 = client.CoreV1Api()
for p in v1.list_namespaced_pod("default").items:
    print(p.metadata.name, p.status.phase)

If you're already running inside a cluster, use config.load_incluster_config() and the service account.

For higher-level cluster work, kr8s and pykube-ng are friendlier than the auto-generated official client.

Calling kubectl via subprocess is also legitimate when you just need to apply a YAML you already have:

subprocess.run(["kubectl","apply","-f","manifest.yaml"], check=True)

Cloud SDKs

AWS — boto3

import boto3

s3 = boto3.client("s3")
s3.upload_file("local.txt", "my-bucket", "remote.txt")
for obj in s3.list_objects_v2(Bucket="my-bucket").get("Contents", []):
    print(obj["Key"], obj["Size"])

ec2 = boto3.client("ec2")
ec2.start_instances(InstanceIds=["i-0abc..."])

GCP

from google.cloud import storage
client = storage.Client()
client.bucket("my-bucket").blob("remote.txt").upload_from_filename("local.txt")

Azure

from azure.identity import DefaultAzureCredential
from azure.storage.blob import BlobServiceClient
cred = DefaultAzureCredential()
svc = BlobServiceClient(account_url="https://x.blob.core.windows.net", credential=cred)

Authentication: rely on the SDK's default credential chain (instance metadata when on-cloud, env vars or local CLI cache otherwise). Don't hard-code keys.

IaC from Python

• Terraform is HCL, but you can drive it from Python with the python-terraform wrapper or just subprocess.run(["terraform","apply", …], check=True). CDKTF (CDK for Terraform) lets you author HCL from Python.
• Pulumi is fully Python — define resources as Python objects.
• Ansible modules are Python; you can call playbooks via ansible-runner from Python code.

# Pulumi snippet
import pulumi
import pulumi_aws as aws

bucket = aws.s3.Bucket("logs", versioning=aws.s3.BucketVersioningArgs(enabled=True))
pulumi.export("bucket_name", bucket.id)

For wrapping an existing Terraform project:

subprocess.run(["terraform","-chdir=infra","init"], check=True)
subprocess.run(["terraform","-chdir=infra","apply","-auto-approve"], check=True)

CI / CD touchpoints

• GitHub Actions: workflow YAML calls python -m pip install -e .[dev] then pytest and mypy.
• GitLab CI: similar, in .gitlab-ci.yml.
• Jenkins: Jenkinsfile with a python stage, or pipenv/poetry/uv step.

A representative GitHub Actions job for a Python tool:

# .github/workflows/test.yml
name: test
on: [push, pull_request]
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
          cache: pip
      - run: python -m pip install -U pip
      - run: python -m pip install -e '.[dev]'
      - run: ruff check .
      - run: mypy src
      - run: pytest --cov=src

For releases, tag-driven: build with python -m build, upload with twine upload, or use python-semantic-release / release-please.

Performance & gotchas

• The GIL means CPU-bound multithreading doesn't help. For CPU work, use multiprocessing or concurrent.futures.ProcessPoolExecutor. For I/O-bound work, threads or asyncio are fine.
• subprocess.run(..., shell=True, …) with f-strings is the most common foot-gun in ops Python. Use a list and avoid shell=True.
• Don't open files without specifying encoding=. Default differs across platforms (UTF-8 on Linux, sometimes cp1252 on Windows).
• os.environ is a dict — but mutating it is process-wide. Use subprocess.run(..., env=…) to scope changes to a child process.
• pip install without a virtualenv on a system Python is the road to pain. Always virtualenv.

Troubleshooting

Recipe collection: subprocess patterns

The single most-used module in ops Python. Memorise these shapes and you'll handle 95% of real cases.

Capture output, raise on failure

import subprocess

def run(cmd, **kw):
    """Run cmd as a list; capture stdout/stderr; raise on non-zero exit."""
    return subprocess.run(
        cmd,
        check=True,
        capture_output=True,
        text=True,
        **kw,
    )

r = run(["git", "rev-parse", "HEAD"])
sha = r.stdout.strip()

check=True raises CalledProcessError on non-zero exit. text=True returns str instead of bytes. capture_output=True collects both streams; if you only need stdout, use stdout=subprocess.PIPE.

Stream output line-by-line as it arrives

import subprocess

with subprocess.Popen(
    ["rsync", "-av", "/src/", "/dst/"],
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,
    text=True,
    bufsize=1,
) as p:
    for line in p.stdout:
        print(line.rstrip())
        # log it, parse it, whatever
if p.returncode != 0:
    raise RuntimeError(f"rsync failed: exit {p.returncode}")

bufsize=1 enables line buffering; without it, output is block-buffered and you see nothing until the child exits. Merging stderr into stdout with stderr=subprocess.STDOUT keeps the lines in original order.

Pipe one command into another (without shell=True)

import subprocess

p1 = subprocess.Popen(["dpkg", "-l"], stdout=subprocess.PIPE)
p2 = subprocess.Popen(["grep", "linux-image"], stdin=p1.stdout,
                      stdout=subprocess.PIPE, text=True)
p1.stdout.close()  # let p1 receive SIGPIPE if p2 exits early
out, _ = p2.communicate()

Or, the often-cleaner shape: capture the first command's output and feed it as input= to the second:

listing = subprocess.run(["dpkg", "-l"], capture_output=True, text=True, check=True).stdout
matches = subprocess.run(["grep", "linux-image"], input=listing,
                          capture_output=True, text=True).stdout

Run with a deadline

import subprocess

try:
    r = subprocess.run(
        ["slow-thing", "--input", path],
        capture_output=True, text=True, timeout=30,
    )
except subprocess.TimeoutExpired as e:
    # e.stdout / e.stderr contain whatever was captured before the kill
    raise RuntimeError(f"slow-thing exceeded 30s") from e

timeout= raises after that many seconds and kills the child process.

Run as a different user / with a clean env

import os, subprocess

env = {
    "PATH": "/usr/local/bin:/usr/bin:/bin",
    "HOME": "/var/lib/myapp",
    "LC_ALL": "C.UTF-8",
}
subprocess.run(
    ["./deploy.sh", "v42"],
    env=env,                     # totally clean environment
    cwd="/opt/myapp",
    user="myapp",                # 3.9+
    group="myapp",
    check=True,
)

Pass an explicit env= to avoid leaking secrets via inherited environment. The user= / group= args (Python 3.9+) drop privileges before exec — useful when the parent runs as root.

Send input to stdin

key = subprocess.run(
    ["openssl", "rand", "-base64", "32"],
    capture_output=True, text=True, check=True,
).stdout.strip()

# pipe a value into another tool's stdin
subprocess.run(
    ["base64", "-d"],
    input="aGVsbG8=\n",
    text=True, check=True,
)

Detached background process

import subprocess

p = subprocess.Popen(
    ["./worker", "--config", "x.toml"],
    stdout=open("/var/log/worker.log", "ab"),
    stderr=subprocess.STDOUT,
    stdin=subprocess.DEVNULL,
    start_new_session=True,        # detach from parent's process group
)
print(f"started pid={p.pid}")
# parent can now exit; child keeps running

Anti-pattern table

Recipe collection: pathlib in practice

pathlib.Path is the modern way to handle paths. It's worth re-learning the patterns explicitly because os.path muscle memory is sticky.

Glob, recursive glob, filtered iteration

from pathlib import Path

base = Path("/var/log")

# Direct children
for p in base.iterdir():
    if p.is_file():
        print(p.name, p.stat().st_size)

# Glob (top level only)
for p in base.glob("*.log"):
    ...

# Recursive glob
for p in base.rglob("*.gz"):
    ...

# Combined predicates
big = [p for p in base.rglob("*") if p.is_file() and p.stat().st_size > 100 * 1024 * 1024]

Read / write text and bytes atomically

from pathlib import Path

p = Path("config.toml")
text = p.read_text(encoding="utf-8")
p.write_text(new_text, encoding="utf-8")

# Atomic write: write to a sibling temp file, fsync, rename
import os, tempfile

def atomic_write_text(path: Path, content: str, encoding="utf-8"):
    path = Path(path)
    fd, tmp = tempfile.mkstemp(dir=path.parent, prefix=path.name, suffix=".tmp")
    try:
        with os.fdopen(fd, "w", encoding=encoding) as f:
            f.write(content)
            f.flush()
            os.fsync(f.fileno())
        os.replace(tmp, path)        # atomic on POSIX and Windows
    except Exception:
        Path(tmp).unlink(missing_ok=True)
        raise

Path manipulation idioms

from pathlib import Path

p = Path("/var/log/nginx/access.log.1.gz")

p.name           # 'access.log.1.gz'
p.stem           # 'access.log.1'    — last dot only
p.suffix         # '.gz'
p.suffixes       # ['.log', '.1', '.gz']
p.parent         # PosixPath('/var/log/nginx')
p.parents[1]     # PosixPath('/var/log')
p.parts          # ('/', 'var', 'log', 'nginx', 'access.log.1.gz')

# Join (use / operator, not concatenation)
out = p.parent / (p.stem + ".rotated.gz")

# Replace a suffix
p.with_suffix(".bak")        # PosixPath('/var/log/nginx/access.log.1.bak')

# Test
p.exists(); p.is_file(); p.is_dir(); p.is_symlink()

# Resolve (absolute, follow symlinks; raises if missing on 3.11+ unless strict=False)
p.resolve(strict=False)

# Relative path
abs_log = Path("/var/log/app/foo.log")
abs_log.relative_to("/var/log")    # PosixPath('app/foo.log')

Cross-platform home and config

from pathlib import Path
import os

# User home
Path.home()

# XDG_CONFIG_HOME with fallback (Linux/macOS); on Windows use APPDATA
def app_config_dir(name: str) -> Path:
    if os.name == "nt":
        return Path(os.environ.get("APPDATA", Path.home() / "AppData" / "Roaming")) / name
    base = Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config"))
    return base / name

cfg_dir = app_config_dir("myapp")
cfg_dir.mkdir(parents=True, exist_ok=True)

Walk a tree with predicates

from pathlib import Path

def walk(root: Path, predicate):
    for p in Path(root).rglob("*"):
        if predicate(p):
            yield p

# All Python files modified in the last 7 days
import time
cutoff = time.time() - 7 * 86400
for p in walk(".", lambda p: p.is_file() and p.suffix == ".py" and p.stat().st_mtime > cutoff):
    print(p)

Recipe collection: HTTP, retries, and rate limits

requests is the de-facto sync HTTP client. httpx is a modern alternative with the same API plus async. Patterns below are written against requests; translate trivially to httpx.

Always set a timeout

import requests

r = requests.get("https://api.example.com/v1/things", timeout=(3.05, 30))
# (connect_timeout, read_timeout). NEVER call requests without timeout.
r.raise_for_status()
data = r.json()

A bare requests.get(url) hangs forever if the server hangs. Make a team rule: every call has a timeout=.

Session with retries and connection pooling

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def make_session(
    retries=5,
    backoff=0.5,
    status_forcelist=(429, 500, 502, 503, 504),
    pool=20,
):
    s = requests.Session()
    retry = Retry(
        total=retries,
        backoff_factor=backoff,
        status_forcelist=status_forcelist,
        allowed_methods=("GET", "PUT", "DELETE", "HEAD", "OPTIONS", "POST"),
        respect_retry_after_header=True,
    )
    adapter = HTTPAdapter(max_retries=retry, pool_connections=pool, pool_maxsize=pool)
    s.mount("http://", adapter)
    s.mount("https://", adapter)
    return s

session = make_session()
session.headers.update({
    "User-Agent": "myapp/1.0 (+ops@example.com)",
    "Authorization": f"Bearer {os.environ['API_TOKEN']}",
})
r = session.get("https://api.example.com/v1/things", timeout=(3, 30))

backoff_factor=0.5 produces sleeps of 0.5, 1, 2, 4, 8 seconds between retries. respect_retry_after_header=True honours Retry-After headers on 429/503 responses.

Stream a large download

import requests
from pathlib import Path

def download(url: str, dst: Path, chunk=64 * 1024):
    with requests.get(url, stream=True, timeout=(5, 60)) as r:
        r.raise_for_status()
        with open(dst, "wb") as f:
            for c in r.iter_content(chunk_size=chunk):
                f.write(c)

Pagination — collect all pages

def paginate(url: str, params=None):
    """Yield records from a paginated API that returns Link headers."""
    while url:
        r = session.get(url, params=params, timeout=(3, 30))
        r.raise_for_status()
        yield from r.json()["items"]
        url = r.links.get("next", {}).get("url")
        params = None    # only on first request

all_items = list(paginate("https://api.example.com/v1/items"))

Token-bucket rate limit

import time, threading

class RateLimiter:
    def __init__(self, rate_per_sec: float, burst: int):
        self.rate = rate_per_sec
        self.capacity = burst
        self.tokens = burst
        self.updated = time.monotonic()
        self.lock = threading.Lock()

    def acquire(self, n=1):
        while True:
            with self.lock:
                now = time.monotonic()
                self.tokens = min(self.capacity, self.tokens + (now - self.updated) * self.rate)
                self.updated = now
                if self.tokens >= n:
                    self.tokens -= n
                    return
                wait = (n - self.tokens) / self.rate
            time.sleep(wait)

limit = RateLimiter(rate_per_sec=10, burst=20)
for item in items:
    limit.acquire()
    session.post(url, json=item, timeout=(3, 30))

Webhook signature verification

import hashlib, hmac, secrets

def verify_sig(secret: bytes, body: bytes, signature_hex: str) -> bool:
    expected = hmac.new(secret, body, hashlib.sha256).hexdigest()
    return secrets.compare_digest(expected, signature_hex)

secrets.compare_digest is the constant-time comparison you must use for any HMAC check; == leaks timing.

Recipe collection: text and structured data

CSV in and out

import csv
from pathlib import Path

# Read with header → list of dicts
with open("input.csv", newline="", encoding="utf-8") as f:
    rows = list(csv.DictReader(f))

# Filter and write
keep = [r for r in rows if int(r["amount"]) > 100]
with open("output.csv", "w", newline="", encoding="utf-8") as f:
    w = csv.DictWriter(f, fieldnames=rows[0].keys())
    w.writeheader()
    w.writerows(keep)

newline="" is required on all platforms. CSV files written without it pick up extra blank lines on Windows.

JSON with sane defaults

import json
from pathlib import Path

# Read
data = json.loads(Path("config.json").read_text(encoding="utf-8"))

# Write — pretty, deterministic, UTF-8
Path("out.json").write_text(
    json.dumps(data, indent=2, sort_keys=True, ensure_ascii=False) + "\n",
    encoding="utf-8",
)

# Decode datetimes / Decimals (they aren't JSON natively)
import datetime, decimal

def default(o):
    if isinstance(o, (datetime.datetime, datetime.date)):
        return o.isoformat()
    if isinstance(o, decimal.Decimal):
        return str(o)
    raise TypeError(f"not JSON-serialisable: {type(o)}")

json.dumps(obj, default=default)

YAML

import yaml

# ALWAYS use safe_load; load() can execute arbitrary Python via tags
data = yaml.safe_load(Path("config.yml").read_text(encoding="utf-8"))

# Dump in block style with stable key order
print(yaml.safe_dump(data, default_flow_style=False, sort_keys=False))

TOML

# 3.11+
import tomllib
data = tomllib.loads(Path("pyproject.toml").read_text(encoding="utf-8"))
# Pre-3.11: pip install tomli, then `import tomli as tomllib`

# Writing TOML: pip install tomli-w
import tomli_w
Path("out.toml").write_bytes(tomli_w.dumps(data).encode("utf-8"))

Templates with Jinja2

from jinja2 import Environment, FileSystemLoader, select_autoescape

env = Environment(
    loader=FileSystemLoader("templates"),
    autoescape=select_autoescape(["html", "xml"]),
    keep_trailing_newline=True,
    trim_blocks=True,
    lstrip_blocks=True,
)
tpl = env.get_template("nginx.conf.j2")
rendered = tpl.render(servers=servers, upstream="api")
Path("/etc/nginx/conf.d/api.conf").write_text(rendered, encoding="utf-8")

autoescape matters when the output is HTML/XML and any input might come from outside your script. Skip it for config files where literal values are wanted.

Recipe collection: pytest patterns

A short collection of fixtures and parametrisation patterns that come up in ops codebases.

tmp_path and capsys

def test_writes_config(tmp_path):
    out = tmp_path / "config.yml"
    write_config(out, {"foo": "bar"})
    assert out.exists()
    assert "foo: bar" in out.read_text(encoding="utf-8")

def test_prints_warning(capsys):
    do_thing_with_warning()
    captured = capsys.readouterr()
    assert "WARNING" in captured.err

Parametrise across cases

import pytest

@pytest.mark.parametrize(
    "input,expected",
    [
        ("",         0),
        ("a",        1),
        ("a,b,c",    3),
        ("a, b, c",  3),     # whitespace tolerant
    ],
    ids=["empty", "single", "no-space", "with-space"],
)
def test_count_csv_fields(input, expected):
    assert count_csv_fields(input) == expected

Fixture with dependency and finaliser

import pytest, subprocess

@pytest.fixture(scope="session")
def docker_pg():
    """Spin up a Postgres container for the test session."""
    name = "test-pg"
    subprocess.run(
        ["docker", "run", "-d", "--rm", "--name", name,
         "-e", "POSTGRES_PASSWORD=test", "-p", "55432:5432",
         "postgres:16-alpine"],
        check=True,
    )
    try:
        wait_for_port("127.0.0.1", 55432, timeout=30)
        yield "postgresql://postgres:test@127.0.0.1:55432/postgres"
    finally:
        subprocess.run(["docker", "rm", "-f", name])

def test_query(docker_pg):
    import psycopg
    with psycopg.connect(docker_pg) as conn:
        assert conn.execute("select 1").fetchone() == (1,)

Mocking subprocess and the clock

from unittest.mock import patch
import subprocess

def test_calls_git(monkeypatch):
    calls = []
    def fake_run(cmd, **kw):
        calls.append(cmd)
        return subprocess.CompletedProcess(cmd, 0, stdout="abc123\n", stderr="")
    monkeypatch.setattr(subprocess, "run", fake_run)

    sha = current_sha()
    assert sha == "abc123"
    assert calls == [["git", "rev-parse", "HEAD"]]

def test_at_specific_time(monkeypatch):
    import datetime
    fixed = datetime.datetime(2026, 1, 1, 12, 0, tzinfo=datetime.timezone.utc)
    class FakeDT(datetime.datetime):
        @classmethod
        def now(cls, tz=None):
            return fixed.astimezone(tz) if tz else fixed
    monkeypatch.setattr("yourmod.datetime", FakeDT)
    assert is_new_year() is True

Skipping conditionally

import os, sys, pytest

@pytest.mark.skipif(sys.platform == "win32", reason="POSIX-only")
def test_chmod():
    ...

@pytest.mark.skipif("CI" not in os.environ, reason="manual-only")
def test_against_real_aws():
    ...

Shared config in conftest.py

# conftest.py at the repo root
import os, pytest

@pytest.fixture(autouse=True)
def _utc_default(monkeypatch):
    """Force UTC for every test so date logic isn't host-timezone-dependent."""
    monkeypatch.setenv("TZ", "UTC")
    import time; time.tzset()

autouse=True applies the fixture to every test in scope without the test having to ask for it. Use sparingly.

Recipe collection: logging, structured logs, and observability

A sane root logger

import logging, sys, os

def configure_logging(level=None):
    level = level or os.environ.get("LOG_LEVEL", "INFO").upper()
    fmt = "%(asctime)s %(levelname)-7s %(name)s %(message)s"
    logging.basicConfig(level=level, format=fmt, datefmt="%Y-%m-%dT%H:%M:%S%z",
                         stream=sys.stderr)
    # Quiet noisy libraries
    logging.getLogger("urllib3").setLevel(logging.WARNING)
    logging.getLogger("botocore").setLevel(logging.WARNING)

configure_logging()
log = logging.getLogger(__name__)
log.info("starting up")

Structured JSON logs

import json, logging, time

class JSONFormatter(logging.Formatter):
    def format(self, record):
        payload = {
            "ts":      time.strftime("%Y-%m-%dT%H:%M:%S%z", time.localtime(record.created)),
            "level":   record.levelname,
            "logger":  record.name,
            "msg":     record.getMessage(),
        }
        # Allow arbitrary extra fields via logger.info("...", extra={"key": "value"})
        for k, v in record.__dict__.items():
            if k in ("msg", "args", "name", "levelname", "levelno", "pathname",
                      "filename", "module", "exc_info", "exc_text", "stack_info",
                      "lineno", "funcName", "created", "msecs", "relativeCreated",
                      "thread", "threadName", "processName", "process", "message"):
                continue
            payload[k] = v
        if record.exc_info:
            payload["exc"] = self.formatException(record.exc_info)
        return json.dumps(payload, default=str)

handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())
logging.basicConfig(handlers=[handler], level=logging.INFO)

log = logging.getLogger(__name__)
log.info("user login", extra={"user_id": 42, "ip": "10.0.0.5"})

Per-request correlation IDs

import contextvars, logging, uuid

_request_id = contextvars.ContextVar("request_id", default="-")

def new_request_id():
    rid = uuid.uuid4().hex[:12]
    _request_id.set(rid)
    return rid

class RequestIdFilter(logging.Filter):
    def filter(self, record):
        record.request_id = _request_id.get()
        return True

logging.getLogger().addFilter(RequestIdFilter())
# Format string can now reference %(request_id)s

Timing helper

import contextlib, logging, time

log = logging.getLogger(__name__)

@contextlib.contextmanager
def timed(label, level=logging.INFO):
    start = time.perf_counter()
    try:
        yield
    finally:
        elapsed_ms = (time.perf_counter() - start) * 1000
        log.log(level, f"{label} took {elapsed_ms:.1f} ms",
                extra={"event": "timed", "label": label, "ms": elapsed_ms})

with timed("fetch users"):
    users = api.fetch_users()

Recipe collection: CLI building blocks

argparse boilerplate

import argparse, sys

def build_parser():
    p = argparse.ArgumentParser(
        prog="myop",
        description="Operate the things.",
    )
    p.add_argument("--config", "-c", default="config.yml", help="config file path")
    p.add_argument("--verbose", "-v", action="count", default=0, help="repeat for more")
    sub = p.add_subparsers(dest="cmd", required=True)

    do = sub.add_parser("deploy", help="deploy a release")
    do.add_argument("version")
    do.add_argument("--target", required=True)
    do.add_argument("--dry-run", action="store_true")

    return p

def main(argv=None):
    args = build_parser().parse_args(argv)
    if args.cmd == "deploy":
        return cmd_deploy(args)
    return 0

if __name__ == "__main__":
    sys.exit(main())

Click — when you want decorators

import click

@click.group()
@click.option("--config", "-c", default="config.yml", show_default=True)
@click.pass_context
def cli(ctx, config):
    ctx.obj = load_config(config)

@cli.command()
@click.argument("version")
@click.option("--target", required=True)
@click.option("--dry-run", is_flag=True)
@click.pass_obj
def deploy(cfg, version, target, dry_run):
    """Deploy VERSION to TARGET."""
    do_deploy(cfg, version, target, dry_run=dry_run)

if __name__ == "__main__":
    cli()

Reading interactive input (with a default and confirm)

def prompt(question: str, default: str | None = None) -> str:
    suffix = f" [{default}]" if default is not None else ""
    while True:
        ans = input(f"{question}{suffix}: ").strip()
        if ans:
            return ans
        if default is not None:
            return default

def confirm(question: str, default=False) -> bool:
    suffix = "[Y/n]" if default else "[y/N]"
    ans = input(f"{question} {suffix} ").strip().lower()
    if not ans:
        return default
    return ans in ("y", "yes")

Reading secrets (no echo)

import getpass
pw = getpass.getpass("Password: ")

Recipe collection: cloud, containers, IaC

boto3 with explicit credentials and a session

import boto3
from botocore.config import Config

session = boto3.Session(profile_name="prod")  # or leave blank for default chain
s3 = session.client(
    "s3",
    region_name="us-east-1",
    config=Config(retries={"mode": "standard", "max_attempts": 5}),
)

# List a bucket, with paginator (handles continuation tokens)
paginator = s3.get_paginator("list_objects_v2")
for page in paginator.paginate(Bucket="my-bucket", Prefix="logs/"):
    for obj in page.get("Contents", []):
        print(obj["Key"], obj["Size"])

Upload with retries and content-type

import mimetypes

def upload(path, bucket, key):
    ct, _ = mimetypes.guess_type(path)
    s3.upload_file(
        Filename=str(path),
        Bucket=bucket,
        Key=key,
        ExtraArgs={"ContentType": ct or "application/octet-stream"},
    )

Presigned URL

url = s3.generate_presigned_url(
    "get_object",
    Params={"Bucket": "my-bucket", "Key": "report.pdf"},
    ExpiresIn=900,
)

Docker SDK — run a container, capture its logs

import docker

client = docker.from_env()
container = client.containers.run(
    "alpine:3.20",
    command=["sh", "-c", "echo hello && sleep 1 && echo world"],
    detach=True,
    remove=False,
)
try:
    rc = container.wait()["StatusCode"]
    logs = container.logs().decode("utf-8")
    print(rc, logs)
finally:
    container.remove(force=True)

Kubernetes client — pod listing, exec, scaling

from kubernetes import client, config

config.load_kube_config()       # ~/.kube/config
# config.load_inclust

…(truncated)