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())

# /// script
# requires-python = ">=3.12"
# dependencies = ["httpx"]
# ///

uv add --script reconcile.py httpx     # writes the block for you
uv run reconcile.py inventory.csv

import json, 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

with subprocess.Popen(cmd, stdout=subprocess.PIPE, text=True) as p:
    for line in p.stdout:
        handle(line)
if p.returncode != 0:
    raise subprocess.CalledProcessError(p.returncode, cmd)

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)     # same filesystem as the target
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            f.write(data)
            f.flush()
            os.fsync(f.fileno())                     # data on disk before the rename
        os.replace(tmp, path)                        # atomic rename on POSIX
    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(),
            **getattr(record, "fields", {}),
        }
        if record.exc_info:
            payload["exc"] = self.formatException(record.exc_info)
        return json.dumps(payload)

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

log.info("deployed", 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

def transient(e: BaseException) -> bool:
    if isinstance(e, httpx.TransportError):          # timeouts, connection resets
        return True
    return isinstance(e, httpx.HTTPStatusError) and e.response.status_code in {429, 502, 503, 504}

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

from pydantic import SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_prefix="APP_", env_file=".env")

    api_url: str
    api_token: SecretStr                          # printed as '**********'
    timeout: float = 10.0
    dry_run: bool = False

settings = Settings()                             # raises ValidationError if APP_API_URL is missing
token = settings.api_token.get_secret_value()     # explicit access to the real value

from dataclasses import dataclass

@dataclass(frozen=True, slots=True)
class Host:
    name: str
    ip: str
    zone: 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 fetch_all(urls: list[str]) -> list[dict]:
    limit = asyncio.Semaphore(20)                  # at most 20 requests in flight
    async with httpx.AsyncClient(timeout=10) as client:
        async def one(u: str) -> dict:
            async with limit:
                r = await client.get(u)
                r.raise_for_status()
                return r.json()
        async with asyncio.TaskGroup() as tg:      # 3.11+: cancels the rest if one fails
            tasks = [tg.create_task(one(u)) for u in urls]
    return [t.result() for t in tasks]

results = asyncio.run(fetch_all(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"

[dependency-groups]
dev = ["pytest>=8", "mypy>=1.10", "ruff"]

[build-system]
requires = ["uv_build>=0.12,<0.13"]      # `uv init` writes a matching range
build-backend = "uv_build"

[tool.ruff]
line-length = 100

[tool.ruff.lint]
select = ["E", "F", "I", "B", "UP", "S"]

[tool.mypy]
strict = true

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

uv sync                            # create .venv if needed, install exactly what uv.lock lists
uv run pytest                      # runs in .venv, syncing first
uv add httpx                       # updates pyproject.toml and uv.lock
uv sync --locked                   # CI: error if uv.lock does not match pyproject.toml
uv sync --frozen                   # use uv.lock as-is without checking it (Docker layers)
uv sync --no-dev                   # production install without the dev group
uv build                           # sdist and wheel into dist/

uv python pin 3.13                      # writes .python-version; uv run and sync use that interpreter
uv python list --only-installed         # interpreters uv knows about
uv lock --upgrade                       # re-resolve everything to the newest allowed versions
uv lock --upgrade-package httpx         # bump one package within its constraint
uv tree                                 # dependency tree from the lock file
uv tree --invert --package certifi      # who depends on certifi
uv add 'httpx>=0.28,<1' --optional cli  # optional extra; install with uv sync --extra cli
uv add --group lint ruff                # a named dependency group; uv sync --group lint
uv sync --all-groups                    # every group, for a full local environment
uv run --with rich python -c 'import rich'   # temporarily add a package without touching the lock
uv run --no-sync pytest                 # skip the sync step when the environment is known good
uv export --format requirements.txt --no-dev -o requirements.txt   # for tools that only read requirements files
uv pip install -r requirements.txt      # pip-compatible interface into the active or project venv
uv cache prune --ci                     # drop pre-built wheels; keeps what CI reuses
uv build && uv publish --token "$PYPI_TOKEN"   # build then upload; use trusted publishing in CI instead of a token

from collections.abc import Iterable, Iterator, Mapping, Sequence, Callable
from typing import Literal, Protocol, TypedDict, NewType, TypeVar, overload, assert_never

def first[T](items: Sequence[T], default: T) -> T:        # 3.12 generic syntax; TypeVar before that
    return items[0] if items else default

Zone = Literal["au-east", "au-west"]                       # only these strings type-check

class Row(TypedDict, total=False):                         # shape of a dict from JSON or csv.DictReader
    name: str
    ip: str
    zone: Zone

class Fetcher(Protocol):                                   # structural: any object with this method satisfies it
    def get(self, url: str, /) -> bytes: ...

HostId = NewType("HostId", int)                            # distinct from int to the checker, an int at runtime

def handle(state: Literal["up", "down"]) -> str:
    match state:
        case "up": return "ok"
        case "down": return "alert"
        case _: assert_never(state)                        # mypy errors if a Literal member is unhandled

[tool.mypy]
python_version = "3.12"
strict = true
warn_unreachable = true
files = ["src", "tests"]

[[tool.mypy.overrides]]
module = ["somelib.*"]
ignore_missing_imports = true

uv run mypy                              # uses [tool.mypy] files
uv run mypy --install-types --non-interactive   # fetch stub packages mypy suggests
uv run mypy --strict --warn-unused-ignores src   # find # type: ignore comments that no longer do anything

from dataclasses import dataclass, field, asdict, replace

@dataclass(frozen=True, slots=True, kw_only=True)         # kw_only: callers must name every field
class Deploy:
    service: str
    version: str
    replicas: int = 2
    tags: tuple[str, ...] = ()
    labels: dict[str, str] = field(default_factory=dict)   # never a mutable default

    def __post_init__(self) -> None:
        if self.replicas < 1:
            raise ValueError("replicas must be >= 1")

d = Deploy(service="api", version="1.4.2")
d2 = replace(d, replicas=4)                                # copy with changes; frozen instances are immutable
asdict(d2)                                                 # nested dict, for json.dumps

from pydantic import BaseModel, Field, field_validator, ValidationError, TypeAdapter

class Item(BaseModel, frozen=True, extra="forbid"):        # extra="forbid": unknown keys are an error
    id: int
    name: str = Field(min_length=1, max_length=64)
    price_cents: int = Field(ge=0)
    tags: list[str] = []

    @field_validator("tags")
    @classmethod
    def lowercase_tags(cls, v: list[str]) -> list[str]:
        return [t.lower() for t in v]

try:
    item = Item.model_validate_json(body)                  # bytes or str straight from the wire
except ValidationError as e:
    log.warning("bad payload: %s", e.errors())             # a list of {loc, msg, type}
    raise

items = TypeAdapter(list[Item]).validate_python(payload)   # validate a plain list without a wrapper model
item.model_dump(mode="json")                               # dict with JSON-safe types (datetimes as strings)

import asyncio

async def main() -> None:
    async with asyncio.timeout(60):                           # 3.11+: cancels the block when exceeded
        data = await asyncio.to_thread(read_big_file, path)   # blocking call in the default executor
        async with asyncio.TaskGroup() as tg:
            t1 = tg.create_task(fetch(data.url), name="fetch")
            t2 = tg.create_task(publish(data), name="publish")
    print(t1.result(), t2.result())

asyncio.run(main())

queue: asyncio.Queue[Job] = asyncio.Queue(maxsize=100)      # bounded: producers block when consumers lag

async def worker(name: str) -> None:
    while True:
        job = await queue.get()
        try:
            await process(job)
        finally:
            queue.task_done()                                 # pairs with queue.join()

async def run(jobs: list[Job]) -> None:
    async with asyncio.TaskGroup() as tg:
        workers = [tg.create_task(worker(f"w{i}")) for i in range(8)]
        for j in jobs:
            await queue.put(j)
        await queue.join()                                    # every put has had a task_done
        for w in workers:
            w.cancel()                                        # TaskGroup absorbs the CancelledError

import logging.config

logging.config.dictConfig({
    "version": 1,
    "disable_existing_loggers": False,                    # keep loggers created at import time
    "formatters": {
        "json": {"()": "myapp.logging.JsonFormatter"},    # "()" names a factory
        "plain": {"format": "%(asctime)s %(levelname)-8s %(name)s: %(message)s"},
    },
    "handlers": {
        "stderr": {"class": "logging.StreamHandler", "formatter": "json", "stream": "ext://sys.stderr"},
        "file": {
            "class": "logging.handlers.RotatingFileHandler",
            "filename": "/var/log/myapp/app.log",
            "maxBytes": 50_000_000, "backupCount": 5, "formatter": "plain",
        },
    },
    "loggers": {
        "httpx": {"level": "WARNING"},                    # quieten a chatty library
        "myapp.db": {"level": "DEBUG"},
    },
    "root": {"level": "INFO", "handlers": ["stderr", "file"]},
})

import pytest

@pytest.fixture
def tmp_config(tmp_path: Path) -> Path:                       # tmp_path is a built-in fixture
    p = tmp_path / "app.yaml"
    p.write_text("timeout: 5\n", encoding="utf-8")
    return p

@pytest.fixture(scope="session")
def db_url() -> Iterator[str]:
    container = start_postgres()                              # once for the whole run
    yield container.url
    container.stop()                                          # teardown after the last test

@pytest.fixture(autouse=True)
def no_network(monkeypatch: pytest.MonkeyPatch) -> None:      # applies to every test in scope
    monkeypatch.delenv("HTTP_PROXY", raising=False)

@pytest.mark.parametrize(
    ("raw", "expected"),
    [
        ("5s", 5.0),
        ("2m", 120.0),
        pytest.param("", None, id="empty"),
        pytest.param("bad", None, marks=pytest.mark.xfail(raises=ValueError, strict=True)),
    ],
)
def test_parse_duration(raw: str, expected: float | None) -> None:
    assert parse_duration(raw) == expected

uv run pytest -k 'parse and not slow'       # expression over test names and markers
uv run pytest --lf                          # only tests that failed last time
uv run pytest -x --pdb                      # drop into the debugger at the first failure
uv run pytest --fixtures tests/             # every fixture available, with docstrings
uv run pytest --setup-show tests/test_x.py  # print fixture setup and teardown order
uv run pytest -p no:cacheprovider -q        # no .pytest_cache, for read-only CI checkouts

# Serve the current directory on localhost only
python3 -m http.server 8000 --bind 127.0.0.1

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

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

# Profile a script, slowest cumulative first
python3 -m cProfile -s cumtime script.py | head -25

# Where a module is loaded from and its version
python3 -c 'import httpx, inspect; print(httpx.__version__, inspect.getfile(httpx))'

# Base64 and URL encoding
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 8601 in UTC (datetime.UTC needs 3.11+)
python3 -c 'import sys,datetime as d; print(d.datetime.fromtimestamp(int(sys.argv[1]), d.UTC).isoformat())' 1700000000

2023-11-14T22:13:20+00:00

# Validate a YAML file (needs PyYAML)
uv run --with pyyaml python -c 'import sys,yaml; yaml.safe_load(open(sys.argv[1]))' config.yaml

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

# Check a wheel's metadata before publishing
uvx twine check dist/*

# Every third-party import in the tree, to compare against pyproject dependencies
grep -rhoE '^(from|import) [a-zA-Z_]+' src | awk '{print $2}' | sort -u

# Run a module's tests with warnings turned into errors
uv run pytest -W error::DeprecationWarning tests/

# Dump every environment variable a Settings class would read
python3 -c 'from myapp.config import Settings; print(Settings.model_json_schema()["properties"].keys())'

# Byte-compile everything to catch syntax errors without importing
python3 -m compileall -q src/

# Show the resolved dependency tree from the lock file
uv tree --depth 1

# Start a REPL with the project importable and a variable pre-loaded
uv run python -i -c 'from myapp.config import Settings; s = Settings()'

import random, time

def retry(fn, *, attempts: int = 5, base: float = 0.5, cap: float = 10.0, retry_on=(OSError,)):
    for i in range(attempts):
        try:
            return fn()
        except retry_on as e:
            if i == attempts - 1:
                raise
            delay = min(cap, base * 2**i) * random.uniform(0.5, 1.5)
            log.warning("attempt %d failed (%s), retrying in %.1fs", i + 1, e, delay)
            time.sleep(delay)

client = httpx.Client(
    base_url="https://api.example.com",
    timeout=httpx.Timeout(connect=3.0, read=15.0, write=10.0, pool=5.0),
    limits=httpx.Limits(max_connections=50, max_keepalive_connections=10),
    transport=httpx.HTTPTransport(retries=2),          # retries connection errors only, not HTTP status codes
    headers={"authorization": f"Bearer {token}"},
)

import signal

stop = False
def _handle(signum, frame):
    global stop
    stop = True
    log.info("signal %s received, finishing current item", signal.Signals(signum).name)

signal.signal(signal.SIGTERM, _handle)
signal.signal(signal.SIGINT, _handle)
for item in work:
    if stop:
        break
    process(item)

async def main() -> None:
    loop = asyncio.get_running_loop()
    task = asyncio.current_task()
    for sig in (signal.SIGTERM, signal.SIGINT):
        loop.add_signal_handler(sig, task.cancel)
    try:
        await serve()
    except asyncio.CancelledError:
        await drain()                                   # bounded cleanup, then let the cancellation finish
        raise

import os, tomllib

DEFAULTS = {"timeout": 10.0, "workers": 4, "api_url": ""}

def load(path: Path) -> dict:
    cfg = dict(DEFAULTS)
    if path.is_file():
        with path.open("rb") as f:                      # tomllib needs binary mode
            cfg.update(tomllib.load(f))
    for k in cfg:
        if (v := os.getenv(f"APP_{k.upper()}")) is not None:
            cfg[k] = type(DEFAULTS[k])(v)               # coerce to the default's type
    return cfg

from concurrent.futures import ThreadPoolExecutor

def run_all(items, fn, workers: int = 8) -> tuple[int, int]:
    ok = failed = 0
    with ThreadPoolExecutor(max_workers=workers) as pool:
        for item, result in zip(items, pool.map(lambda i: _safe(fn, i), items)):
            if isinstance(result, Exception):
                failed += 1
                log.error("item %s failed: %s", item, result)
            else:
                ok += 1
    return ok, failed

def _safe(fn, item):
    try:
        return fn(item)
    except Exception as e:                              # captured, not raised, so the pool keeps going
        return e

from contextlib import contextmanager
import time

@contextmanager
def timed(label: str):
    start = time.perf_counter()
    try:
        yield
    finally:
        log.info("%s took %.3fs", label, time.perf_counter() - start)

with timed("reconcile"):
    reconcile()

import contextvars, logging

request_id: contextvars.ContextVar[str] = contextvars.ContextVar("request_id", default="-")

class ContextFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.request_id = request_id.get()            # available to formatters as %(request_id)s
        return True

logging.getLogger().addFilter(ContextFilter())
token = request_id.set(incoming_id)                     # per request; asyncio tasks inherit a copy

import hashlib

def download(client: httpx.Client, url: str, dest: Path, sha256: str) -> None:
    h = hashlib.sha256()
    tmp = dest.with_suffix(dest.suffix + ".part")
    with client.stream("GET", url) as r, tmp.open("wb") as f:
        r.raise_for_status()
        for chunk in r.iter_bytes(1 << 20):
            f.write(chunk)
            h.update(chunk)
    if h.hexdigest() != sha256:
        tmp.unlink()
        raise ValueError(f"checksum mismatch for {url}")
    tmp.replace(dest)

def iter_pages(client: httpx.Client, path: str, **params):
    cursor = None
    while True:
        r = client.get(path, params={**params, "cursor": cursor})
        r.raise_for_status()
        body = r.json()
        yield from body["items"]
        cursor = body.get("next_cursor")
        if not cursor:
            return

@pytest.mark.parametrize(
    ("status", "body", "expected"),
    [(200, {"items": [1, 2]}, [1, 2]), (404, {}, []), (500, {}, None)],
)
def test_fetch_items(status: int, body: dict, expected):
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(status, json=body)

    client = httpx.Client(transport=httpx.MockTransport(handler))
    if expected is None:
        with pytest.raises(httpx.HTTPStatusError):
            fetch_items(client)
    else:
        assert fetch_items(client) == expected

[tool.ruff]
line-length = 100
target-version = "py312"
src = ["src", "tests"]

[tool.ruff.lint]
select = ["E", "W", "F", "I", "B", "UP", "S", "PTH", "RUF", "SIM", "ASYNC", "LOG", "T20"]
ignore = ["S101"]                      # assert is fine in tests, and tests are where it appears
per-file-ignores = { "tests/**" = ["S", "T20"] }

[tool.ruff.lint.isort]
known-first-party = ["myapp"]

[tool.ruff.format]
docstring-code-format = true

def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(prog="ops")
    sub = ap.add_subparsers(dest="cmd", required=True)
    p = sub.add_parser("sync", help="reconcile inventory"); p.add_argument("path", type=Path); p.set_defaults(fn=cmd_sync)
    p = sub.add_parser("report", help="print a summary"); p.add_argument("--json", action="store_true"); p.set_defaults(fn=cmd_report)
    args = ap.parse_args(argv)
    return args.fn(args)