cli/docs/api-reference.md

layout, title, description, parent, toc

layout	title	description	parent	toc
default	API Reference	Comprehensive API documentation for all Sunbeam modules	index	true

API Reference

This section provides comprehensive API documentation for all Sunbeam modules, including function signatures, parameters, return values, and usage examples.

CLI Module (sunbeam.cli)

Main Function

def main() -> None

Description: Main entry point for the Sunbeam CLI. Parses arguments and dispatches to appropriate command handlers.

Environment Variables:

• SUNBEAM_SSH_HOST: Required for production environment

Global Arguments:

• --env: Target environment (local or production)
• --context: kubectl context override
• --domain: Domain suffix for production
• --email: ACME email for cert-manager

Command Dispatch

The main function uses a dispatch pattern to route commands to their handlers:

# Example dispatch pattern
if args.verb == "up":
    from sunbeam.cluster import cmd_up
    cmd_up()
elif args.verb == "status":
    from sunbeam.services import cmd_status
    cmd_status(args.target)
# ... additional commands

Kubernetes Module (sunbeam.kube)

Context Management

def set_context(ctx: str, ssh_host: str = "") -> None

Parameters:

• ctx: Kubernetes context name
• ssh_host: SSH host for production tunnel (optional)

Description: Sets the active kubectl context and SSH host for production environments.

def context_arg() -> str

Returns: --context=<active> string for subprocess calls

Kubernetes Operations

def kube(*args, input=None, check=True) -> subprocess.CompletedProcess

Parameters:

• *args: kubectl arguments
• input: stdin input (optional)
• check: Whether to raise exception on non-zero exit (default: True)

Returns: CompletedProcess result

Description: Run kubectl against active context with automatic SSH tunnel management.

def kube_out(*args) -> str

Parameters:

• *args: kubectl arguments

Returns: stdout output (empty string on failure)

def kube_ok(*args) -> bool

Parameters:

• *args: kubectl arguments

Returns: True if command succeeds

def kube_apply(manifest: str, *, server_side: bool = True) -> None

Parameters:

• manifest: YAML manifest string
• server_side: Use server-side apply (default: True)

Description: Apply YAML manifest via kubectl.

Resource Management

def ns_exists(ns: str) -> bool

Parameters:

• ns: Namespace name

Returns: True if namespace exists

def ensure_ns(ns: str) -> None

Parameters:

• ns: Namespace name

Description: Ensure namespace exists, creating if necessary.

def create_secret(ns: str, name: str, **literals) -> None

Parameters:

• ns: Namespace name
• name: Secret name
• **literals: Key-value pairs for secret data

Description: Create or update Kubernetes secret idempotently.

Kustomize Integration

def kustomize_build(overlay: Path, domain: str, email: str = "") -> str

Parameters:

• overlay: Path to kustomize overlay directory
• domain: Domain suffix for substitution
• email: ACME email for substitution (optional)

Returns: Built manifest YAML with substitutions applied

Description: Build kustomize overlay with domain/email substitution and helm support.

SSH Tunnel Management

def ensure_tunnel() -> None

Description: Open SSH tunnel to production cluster if needed. Automatically checks if tunnel is already open.

Utility Functions

def parse_target(s: str | None) -> tuple[str | None, str | None]

Parameters:

• s: Target string in format namespace/name or namespace

Returns: Tuple of (namespace, name)

Examples:

parse_target("ory/kratos")    # ("ory", "kratos")
parse_target("lasuite")       # ("lasuite", None)
parse_target(None)             # (None, None)

def domain_replace(text: str, domain: str) -> str

Parameters:

• text: Input text containing DOMAIN_SUFFIX placeholders
• domain: Domain to substitute

Returns: Text with DOMAIN_SUFFIX replaced by domain

def get_lima_ip() -> str

Returns: IP address of Lima VM

Description: Discover Lima VM IP using multiple fallback methods.

def get_domain() -> str

Returns: Active domain from cluster state

Description: Discover domain from various cluster sources with fallback to Lima IP.

Command Functions

def cmd_k8s(kubectl_args: list[str]) -> int

Parameters:

• kubectl_args: Arguments to pass to kubectl

Returns: Exit code

Description: Transparent kubectl passthrough for active context.

def cmd_bao(bao_args: list[str]) -> int

Parameters:

• bao_args: Arguments to pass to bao CLI

Returns: Exit code

Description: Run bao CLI inside OpenBao pod with root token injection.

Manifests Module (sunbeam.manifests)

Main Command

def cmd_apply(env: str = "local", domain: str = "", email: str = "", namespace: str = "") -> None

Parameters:

• env: Environment (local or production)
• domain: Domain suffix (required for production)
• email: ACME email for cert-manager
• namespace: Limit apply to specific namespace (optional)

Description: Build kustomize overlay, apply domain/email substitution, and apply manifests with convergence handling.

Cleanup Functions

def pre_apply_cleanup(namespaces=None) -> None

Parameters:

• namespaces: List of namespaces to clean (default: all managed namespaces)

Description: Clean up immutable resources and stale secrets before apply.

ConfigMap Management

def _snapshot_configmaps() -> dict

Returns: Dictionary of {ns/name: resourceVersion} for all ConfigMaps

def _restart_for_changed_configmaps(before: dict, after: dict) -> None

Parameters:

• before: ConfigMap snapshot before apply
• after: ConfigMap snapshot after apply

Description: Restart deployments that mount changed ConfigMaps.

Webhook Management

def _wait_for_webhook(ns: str, svc: str, timeout: int = 120) -> bool

Parameters:

• ns: Namespace of webhook service
• svc: Service name
• timeout: Maximum wait time in seconds

Returns: True if webhook becomes ready

Description: Wait for webhook service endpoint to become available.

Utility Functions

def _filter_by_namespace(manifests: str, namespace: str) -> str

Parameters:

• manifests: Multi-document YAML string
• namespace: Namespace to filter by

Returns: Filtered YAML containing only resources for specified namespace

Description: Filter manifests by namespace using string matching.

Services Module (sunbeam.services)

Status Command

def cmd_status(target: str | None) -> None

Parameters:

• target: Optional target in format namespace or namespace/name

Description: Show pod health across managed namespaces with VSO sync status.

Log Command

def cmd_logs(target: str, follow: bool) -> None

Parameters:

• target: Target in format namespace/name
• follow: Whether to stream logs

Description: Stream logs for a service using label-based pod selection.

Get Command

def cmd_get(target: str, output: str = "yaml") -> None

Parameters:

• target: Target in format namespace/name
• output: Output format (yaml, json, or wide)

Description: Get raw kubectl output for a resource.

Restart Command

def cmd_restart(target: str | None) -> None

Parameters:

• target: Optional target in format namespace or namespace/name

Description: Rolling restart of services. If no target, restarts all managed services.

VSO Sync Status

def _vso_sync_status() -> None

Description: Print VaultStaticSecret and VaultDynamicSecret sync health status.

Tools Module (sunbeam.tools)

Tool Management

def ensure_tool(name: str) -> Path

Parameters:

• name: Tool name (must be in TOOLS dictionary)

Returns: Path to executable

Raises:

• ValueError: Unknown tool
• RuntimeError: SHA256 verification failure

Description: Ensure tool is available, downloading and verifying if necessary.

def run_tool(name: str, *args, **kwargs) -> subprocess.CompletedProcess

Parameters:

• name: Tool name
• *args: Arguments to pass to tool
• **kwargs: Additional subprocess.run parameters

Returns: CompletedProcess result

Description: Run tool, ensuring it's downloaded first. Handles PATH setup for kustomize/helm integration.

Internal Utilities

def _sha256(path: Path) -> str

Parameters:

• path: Path to file

Returns: SHA256 hash of file contents

Description: Calculate SHA256 hash (internal utility).

Output Module (sunbeam.output)

Output Functions

def step(msg: str) -> None

Parameters:

• msg: Message to display

Description: Print step header with ==> prefix.

def ok(msg: str) -> None

Parameters:

• msg: Message to display

Description: Print success/info message with indentation.

def warn(msg: str) -> None

Parameters:

• msg: Warning message

Description: Print warning to stderr.

def die(msg: str) -> None

Parameters:

• msg: Error message

Description: Print error to stderr and exit with code 1.

def table(rows: list[list[str]], headers: list[str]) -> str

Parameters:

• rows: List of row data
• headers: List of column headers

Returns: Formatted table string

Description: Generate aligned text table with automatic column sizing.

Usage Patterns

Typical Command Flow

# CLI entry point
from sunbeam.cli import main
main()

# Module usage example
from sunbeam.kube import kube, kube_out, set_context
from sunbeam.output import step, ok, warn

# Set context
set_context("sunbeam")

# Check cluster status
step("Checking cluster status...")
if kube_ok("get", "nodes"):
    ok("Cluster is ready")
    node_count = kube_out("get", "nodes", "--no-headers").count("\n") + 1
    ok(f"Found {node_count} nodes")
else:
    warn("Cluster not ready")

Error Handling Pattern

from sunbeam.output import step, ok, die
from sunbeam.kube import kube_ok

step("Validating cluster requirements...")

# Check required components
required = [
    ("kube-system", "kube-dns"),
    ("ingress-nginx", "ingress-nginx-controller")
]

for ns, deployment in required:
    if not kube_ok("get", "deployment", deployment, "-n", ns):
        die(f"Required deployment {ns}/{deployment} not found")

ok("All requirements satisfied")

Tool Usage Pattern

from sunbeam.tools import ensure_tool, run_tool

# Ensure kubectl is available
kubectl_path = ensure_tool("kubectl")

# Run kubectl command
result = run_tool("kubectl", "get", "pods", "-A", 
                  capture_output=True, text=True)
print(result.stdout)