# Python Devops

> Python reference for ops teams — shelling out, file manipulation, structured data parsing, CLI tools, testing, cloud SDKs, Kubernetes API, IaC and CI/CD patterns.

- Skill: `paulpas/python-devops` (Agent Skill)
- Install (CLI): `npx skillmds@latest add paulpas/python-devops`
- Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/python-devops/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: DevOps & Infra
- License: MIT
- Author: paulpas (https://skillmd.com/u/paulpas)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/paulpas/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):

```toml
[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

```bash
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.

```python
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.

```python
# 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`:

```python
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.

```python
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:

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

---

## Text and regex

```python
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:

```python
"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

```python
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.

```python
# 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()
```

```python
# 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().

```python
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`.

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

---

## Testing with pytest

```python
# 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:

```bash
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:

```python
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

```python
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):

```python
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.

```python
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

```python
# 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:

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

---

## Cloud SDKs

### AWS — boto3

```python
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

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

### Azure

```python
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.

```python
# 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:

```python
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:

```yaml
# .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

| Symptom | Likely cause | Fix |
| --- | --- | --- |
| `ModuleNotFoundError` after install | wrong interpreter | `python -m pip install …` (use the python you'll run with) |
| Tests pass locally, fail in CI | env-dependent path / time / locale | pin `LANG=C.UTF-8`, freeze the timezone, mock time |
| `subprocess` hangs | child waiting on stdin | pass `stdin=subprocess.DEVNULL` or `input=…` |
| `requests.get` hangs forever | no `timeout` | always set `timeout=` |
| YAML loading throws on `!!` tags | safe_load rejects custom tags | good — use `safe_load` and avoid them |
| Import works in REPL, not in script | `cwd` differs; package not on `PYTHONPATH` | install editable: `pip install -e .` |

---

## 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

```python
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

```python
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`)

```python
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:

```python
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

```python
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

```python
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

```python
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

```python
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

| Pattern | Problem | Fix |
| --- | --- | --- |
| `subprocess.run(f"grep {pat} {file}", shell=True)` | shell injection | use a list: `["grep", pat, file]` |
| `subprocess.call(...)` then check return code by hand | misses errors silently | use `run(..., check=True)` |
| Forgetting `text=True` | get `bytes`, accidentally compare to `str` | always set `text=True` for ops scripts |
| No `timeout=` on calls to remote / network tools | hangs forever | set a sensible deadline always |
| Reading `stdout` after `wait()` | deadlocks on large outputs | use `communicate()` or stream as it runs |

---

## 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

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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

```python
# 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

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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`

```python
# 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

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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)

```python
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)

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

---

## Recipe collection: cloud, containers, IaC

### boto3 with explicit credentials and a session

```python
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

```python
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

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

### Docker SDK — run a container, capture its logs

```python
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

```python
from kubernetes import client, config

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

…(truncated)