Python

#!/usr/bin/env python3
"""Reconcile inventory against the API."""
from __future__ import annotations

import argparse
import logging
import sys
from pathlib import Path

log = logging.getLogger("reconcile")


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("inventory", type=Path)
    ap.add_argument("--dry-run", action="store_true")
    ap.add_argument("-v", "--verbose", action="count", default=0)
    args = ap.parse_args()

    logging.basicConfig(
        level=logging.DEBUG if args.verbose else logging.INFO,
        format="%(asctime)s %(levelname)s %(name)s %(message)s",
        stream=sys.stderr,
    )

    if not args.inventory.is_file():
        log.error("inventory not found: %s", args.inventory)
        return 2

    log.info("reconciling %s dry_run=%s", args.inventory, args.dry_run)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

import shlex, subprocess

res = subprocess.run(
    ["kubectl", "get", "pods", "-o", "json"],
    check=True, capture_output=True, text=True, timeout=30,
)
pods = json.loads(res.stdout)

try:
    subprocess.run(cmd, check=True, capture_output=True, text=True, timeout=60)
except subprocess.CalledProcessError as e:
    log.error("command failed rc=%s stderr=%s", e.returncode, e.stderr.strip())
    raise
except subprocess.TimeoutExpired:
    log.error("command timed out: %s", shlex.join(cmd))
    raise

from pathlib import Path

base = Path("/srv/app")
cfg = base / "conf" / "app.yaml"
cfg.exists(); cfg.stat().st_size; cfg.read_text(encoding="utf-8")
list(base.rglob("*.log"))
base.mkdir(parents=True, exist_ok=True)

import os, tempfile

def write_atomic(path: Path, data: str) -> None:
    fd, tmp = tempfile.mkstemp(dir=path.parent)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            f.write(data)
            f.flush()
            os.fsync(f.fileno())
        os.replace(tmp, path)          # atomic on the same filesystem
    except BaseException:
        os.unlink(tmp)
        raise

import json, logging

class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "ts": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"),
            "level": record.levelname,
            "logger": record.name,
            "msg": record.getMessage(),
        }
        if record.exc_info:
            payload["exc"] = self.formatException(record.exc_info)
        payload.update(getattr(record, "extra_fields", {}))
        return json.dumps(payload)

log.info("deployed", extra={"extra_fields": {"service": "api", "version": "1.4.2"}})

import httpx

with httpx.Client(timeout=10.0, headers={"user-agent": "reconcile/1.0"}) as client:
    r = client.get("https://api.example.com/items", params={"limit": 100})
    r.raise_for_status()
    items = r.json()

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=0.5, max=10),
    retry=retry_if_exception_type((httpx.TimeoutException, httpx.HTTPStatusError)),
    reraise=True,
)
def fetch(client: httpx.Client, url: str) -> dict:
    r = client.get(url)
    r.raise_for_status()
    return r.json()

from pydantic import Field
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    api_url: str
    api_token: str = Field(repr=False)          # kept out of repr and logs
    timeout: float = 10.0
    dry_run: bool = False

    model_config = {"env_prefix": "APP_", "env_file": ".env"}

settings = Settings()                            # raises at start-up if anything is missing

from dataclasses import dataclass, field

@dataclass(frozen=True, slots=True)
class Host:
    name: str
    ip: str
    tags: tuple[str, ...] = ()

from collections import Counter, defaultdict

counts = Counter(h.tags[0] for h in hosts if h.tags)
by_zone: dict[str, list[Host]] = defaultdict(list)
for h in hosts:
    by_zone[h.zone].append(h)

import csv
with open("hosts.csv", newline="", encoding="utf-8") as f:
    rows = list(csv.DictReader(f))

def read_events(path: Path):
    with path.open(encoding="utf-8") as f:
        for line in f:                 # one line at a time, not the whole file
            yield json.loads(line)

from concurrent.futures import ThreadPoolExecutor, as_completed

with ThreadPoolExecutor(max_workers=8) as pool:
    futures = {pool.submit(check_host, h): h for h in hosts}
    for fut in as_completed(futures):
        host = futures[fut]
        try:
            result = fut.result()
        except Exception:
            log.exception("check failed host=%s", host.name)

import asyncio, httpx

async def main(urls: list[str]) -> list[dict]:
    limits = asyncio.Semaphore(20)
    async with httpx.AsyncClient(timeout=10) as client:
        async def one(u: str) -> dict:
            async with limits:
                r = await client.get(u)
                r.raise_for_status()
                return r.json()
        return await asyncio.gather(*(one(u) for u in urls))

[project]
name = "reconcile"
version = "1.4.2"
requires-python = ">=3.12"
dependencies = ["httpx>=0.27", "pydantic-settings>=2"]

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

[tool.ruff]
line-length = 100

[tool.mypy]
strict = true

[tool.pytest.ini_options]
addopts = "-q --strict-markers"

uv venv && uv sync                 # resolve and install from the lock file
uv run pytest
uv add httpx                       # updates pyproject.toml and the lock
uv sync --frozen                   # CI: fail if the lock is out of date
pipx install ./dist/*.whl          # install a tool without polluting the environment

# HTTP server for the current directory
python3 -m http.server 8000 --bind 127.0.0.1

# Pretty-print JSON
python3 -m json.tool < data.json

# Time a snippet properly
python3 -m timeit -s 'import json' 'json.dumps({"a":1})'

# What is slow
python3 -m cProfile -s cumtime script.py | head -25

# Inspect a module's location and version
python3 -c 'import httpx, inspect; print(httpx.__version__, inspect.getfile(httpx))'

# Base64 and URL encoding without leaving the shell
python3 -c 'import base64,sys; print(base64.b64encode(sys.stdin.buffer.read()).decode())'
python3 -c 'import urllib.parse,sys; print(urllib.parse.quote(sys.argv[1]))' 'a b&c'

# Epoch to ISO
python3 -c 'import sys,datetime as d; print(d.datetime.fromtimestamp(int(sys.argv[1]), d.UTC).isoformat())' 1700000000

# Validate YAML or JSON in CI
python3 -c 'import sys,yaml; yaml.safe_load(open(sys.argv[1]))' config.yaml

# Which package owns an import
python3 -c 'import importlib.metadata as m; print(m.packages_distributions()["yaml"])'

# Show effective settings from environment
python3 -c 'from app.settings import Settings; print(Settings().model_dump(exclude={"api_token"}))'

# Find the slowest tests
pytest --durations=10 -q

# Check a wheel before publishing
python3 -m twine check dist/*