Helm & Helmfile Automation

import subprocess
import json

def helm(command: list[str], namespace: str | None = None) -> dict | str:
    """Run a helm command and return the result."""
    cmd = ["helm"] + command
    if namespace:
        cmd.extend(["-n", namespace])

    result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
    if result.returncode != 0:
        raise Exception(f"helm failed: {result.stderr}")

    return result.stdout

# List releases
output = helm(["list", "-o", "json"])
releases = json.loads(output)
for release in releases:
    print(f"{release['name']}: {release['status']} (rev {release['revision']})")

# Install a chart
helm(["install", "my-release", "bitnginx/nginx", "--values", "values.yaml"])

# Upgrade a release
helm(["upgrade", "my-release", "bitnginx/nginx", "--values", "values-new.yaml"])

# Uninstall a release
helm(["uninstall", "my-release"])

import yaml
from pathlib import Path
from pydantic import BaseModel, Field

class IngressConfig(BaseModel):
    enabled: bool = False
    host: str = ""
    tls: bool = False

class ResourcesConfig(BaseModel):
    requests_cpu: str = "100m"
    requests_memory: str = "128Mi"
    limits_cpu: str = "500m"
    limits_memory: str = "512Mi"

class HelmValues(BaseModel):
    replica_count: int = Field(default=1, ge=1, le=100)
    image_repository: str
    image_tag: str = "latest"
    ingress: IngressConfig = IngressConfig()
    resources: ResourcesConfig = ResourcesConfig()

def load_and_validate_values(path: Path) -> HelmValues:
    """Load and validate Helm values file."""
    with open(path) as f:
        raw = yaml.safe_load(f)
    return HelmValues(**raw)

# Usage
values = load_and_validate_values(Path("values.yaml"))
print(f"Image: {values.image_repository}:{values.image_tag}")
print(f"Replicas: {values.replica_count}")

import yaml
from pathlib import Path
from datetime import datetime

def generate_values(
    env: str,
    replicas: int,
    image_tag: str,
    enable_monitoring: bool = True,
) -> dict:
    """Generate Helm values based on environment."""
    base = {
        "replica_count": replicas,
        "image": {"repository": "myapp", "tag": image_tag},
        "service": {"type": "ClusterIP", "port": 80},
        "ingress": {"enabled": env != "development"},
        "resources": {
            "requests": {"cpu": "100m", "memory": "128Mi"},
            "limits": {"cpu": "500m", "memory": "512Mi"},
        },
    }

    if env == "production":
        base["replica_count"] = max(replicas, 3)
        base["resources"]["requests"]["cpu"] = "250m"
        base["resources"]["requests"]["memory"] = "256Mi"

    if enable_monitoring:
        base["monitoring"] = {"enabled": True, "port": 9090}

    return base

# Generate values for each environment
for env in ["development", "staging", "production"]:
    values = generate_values(env, replicas=2, image_tag="v1.2.3")
    Path(f"values-{env}.yaml").write_text(yaml.dump(values, default_flow_style=False))

import subprocess
import yaml
from pathlib import Path

def helmfile(command: str, environment: str = "default", **kwargs) -> str:
    """Run a helmfile command."""
    cmd = ["helmfile", "-e", environment, command]
    for key, value in kwargs.items():
        cmd.extend([f"--{key}", str(value)])

    result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
    if result.returncode != 0:
        raise Exception(f"helmfile failed: {result.stderr}")
    return result.stdout

# Sync all releases for an environment
output = helmfile("sync", environment="production")
print(output)

# Diff before applying
output = helmfile("diff", environment="staging")
print(output)

# Template rendering (dry run)
output = helmfile("template", environment="production")

import subprocess
import json
import yaml

def template_chart(chart_path: str, values_file: str | None = None) -> list[dict]:
    """Template a Helm chart and return the rendered manifests."""
    cmd = ["helm", "template", "test-release", chart_path]
    if values_file:
        cmd.extend(["--values", values_file])

    result = subprocess.run(cmd, capture_output=True, text=True)
    if result.returncode != 0:
        raise Exception(f"helm template failed: {result.stderr}")

    # Split YAML documents
    docs = []
    for doc in yaml.safe_load_all(result.stdout):
        if doc:
            docs.append(doc)
    return docs

def validate_chart(chart_path: str) -> bool:
    """Run helm lint on a chart."""
    result = subprocess.run(
        ["helm", "lint", chart_path],
        capture_output=True, text=True,
    )
    print(result.stdout)
    return result.returncode == 0

# Lint the chart
is_valid = validate_chart("./charts/myapp")
print(f"Chart valid: {is_valid}")

# Template and validate
manifests = template_chart("./charts/myapp", "values-test.yaml")
for m in manifests:
    print(f"{m['kind']}: {m['metadata']['name']}")

import subprocess
import json
from datetime import datetime

def get_release_history(release: str, namespace: str = "default") -> list[dict]:
    """Get the revision history of a Helm release."""
    cmd = ["helm", "history", release, "-n", namespace, "-o", "json"]
    result = subprocess.run(cmd, capture_output=True, text=True)
    return json.loads(result.stdout) if result.stdout else []

def rollback_release(release: str, revision: int, namespace: str = "default"):
    """Roll back a Helm release to a specific revision."""
    cmd = ["helm", "rollback", release, str(revision), "-n", namespace]
    result = subprocess.run(cmd, capture_output=True, text=True)
    if result.returncode != 0:
        raise Exception(f"Rollback failed: {result.stderr}")
    print(f"Rolled back {release} to revision {revision}")

# Check history before rollback
history = get_release_history("my-release")
for rev in history:
    print(f"Rev {rev['revision']}: {rev['status']} - {rev.get('description', '')}")

# Rollback to the previous revision
rollback_release("my-release", revision=2)