Python Automation Basics

python3 -m venv .venv
source .venv/bin/activate

pip install requests pyyaml kubernetes     # common SRE deps
pip freeze > requirements.txt

# Run without activating
.venv/bin/python script.py --flag value

#!/usr/bin/env python3
"""One-line description of what this script does."""

import argparse
import logging
import os
import sys

# Configure logging to stderr; caller/CI captures stdout cleanly.
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s — %(message)s",
    stream=sys.stderr,
)
log = logging.getLogger(__name__)


def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument("--namespace", "-n", default="default", help="Kubernetes namespace")
    p.add_argument("--dry-run", action="store_true", help="Print actions without executing")
    p.add_argument("--verbose", "-v", action="store_true", help="Enable debug logging")
    return p.parse_args()


def main() -> int:
    args = parse_args()
    if args.verbose:
        logging.getLogger().setLevel(logging.DEBUG)

    # Prefer env vars for secrets/config; CLI args for runtime overrides.
    token = os.environ.get("API_TOKEN")
    if not token:
        log.error("API_TOKEN environment variable is not set")
        return 1

    log.info("Starting in namespace=%s dry_run=%s", args.namespace, args.dry_run)

    # --- main logic here ---

    return 0


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

import json
import yaml  # pip install pyyaml
from pathlib import Path

# JSON read / write
def load_json(path: str) -> dict:
    return json.loads(Path(path).read_text())

def save_json(data: dict, path: str, pretty: bool = True) -> None:
    indent = 2 if pretty else None
    Path(path).write_text(json.dumps(data, indent=indent, default=str))

# YAML read / write
def load_yaml(path: str) -> dict:
    return yaml.safe_load(Path(path).read_text())

def save_yaml(data: dict, path: str) -> None:
    Path(path).write_text(yaml.dump(data, default_flow_style=False))

# Example: read and patch a Kubernetes manifest
manifest = load_yaml("deployment.yaml")
manifest["spec"]["replicas"] = 3
manifest["spec"]["template"]["spec"]["containers"][0]["image"] = "myapp:v2.0"
save_yaml(manifest, "deployment-patched.yaml")

# Read all YAML docs from a multi-doc file (--- separator)
def load_all_yaml(path: str):
    return list(yaml.safe_load_all(Path(path).read_text()))

# Iterate log file lines efficiently (no full load into memory)
def tail_log(path: str, n: int = 100):
    lines = Path(path).read_text().splitlines()
    return lines[-n:]

import logging
import sys

log = logging.getLogger(__name__)


def safe_open(path: str) -> str:
    """Return file content or raise with a helpful message."""
    try:
        with open(path) as f:
            return f.read()
    except FileNotFoundError:
        log.error("Config file not found: %s", path)
        raise
    except PermissionError:
        log.error("No read permission for: %s", path)
        raise


def with_retries(func, attempts: int = 3, delay: float = 2.0):
    """Retry func up to attempts times with exponential backoff."""
    import time
    last_exc = None
    for attempt in range(1, attempts + 1):
        try:
            return func()
        except Exception as exc:
            last_exc = exc
            wait = delay * (2 ** (attempt - 1))
            log.warning("Attempt %d/%d failed: %s. Retrying in %.1fs", attempt, attempts, exc, wait)
            if attempt < attempts:
                time.sleep(wait)
    raise RuntimeError(f"All {attempts} attempts failed") from last_exc


# Context managers for cleanup
from contextlib import contextmanager

@contextmanager
def temp_kubeconfig(config_data: str):
    """Write a temporary kubeconfig and clean up after use."""
    import tempfile, os
    with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f:
        f.write(config_data)
        tmp_path = f.name
    old = os.environ.get("KUBECONFIG")
    os.environ["KUBECONFIG"] = tmp_path
    try:
        yield tmp_path
    finally:
        os.environ.pop("KUBECONFIG", None)
        if old:
            os.environ["KUBECONFIG"] = old
        os.unlink(tmp_path)

import os
from dataclasses import dataclass


@dataclass
class Config:
    api_url: str
    api_token: str
    namespace: str
    dry_run: bool


def load_config() -> Config:
    """Load config from environment variables with clear error messages."""
    missing = []
    api_url   = os.environ.get("API_URL")   or missing.append("API_URL")
    api_token = os.environ.get("API_TOKEN") or missing.append("API_TOKEN")
    if missing:
        raise EnvironmentError(f"Required env vars not set: {', '.join(str(m) for m in missing)}")
    return Config(
        api_url=api_url,
        api_token=api_token,
        namespace=os.environ.get("NAMESPACE", "default"),
        dry_run=os.environ.get("DRY_RUN", "false").lower() in ("1", "true", "yes"),
    )