Files
beanflows/src/materia/cli.py
Deeman c1d00dcdc4 Refactor to local-first architecture on Hetzner NVMe
Remove distributed R2/Iceberg/SSH pipeline architecture in favor of
local subprocess execution with NVMe storage. Landing data backed up
to R2 via rclone timer.

- Strip Iceberg catalog, httpfs, boto3, paramiko, prefect, pyarrow
- Pipelines run via subprocess.run() with bounded timeouts
- Extract writes to {LANDING_DIR}/psd/{year}/{month}/{etag}.csv.gzip
- SQLMesh reads LANDING_DIR variable, writes to DUCKDB_PATH
- Delete unused provider stubs (ovh, scaleway, oracle)
- Add rclone systemd timer for R2 backup every 6h
- Update supervisor to run pipelines with env vars

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-18 19:50:19 +01:00

159 lines
4.3 KiB
Python

"""Materia CLI - Management interface for BeanFlows.coffee infrastructure."""
from typing import Annotated
import typer
app = typer.Typer(
name="materia",
help="BeanFlows.coffee data platform management CLI",
no_args_is_help=True,
)
@app.command()
def version():
"""Show Materia version."""
typer.echo("Materia CLI v0.1.0")
worker_app = typer.Typer(help="Manage worker instances")
app.add_typer(worker_app, name="worker")
@worker_app.command("list")
def worker_list(
provider: Annotated[str, typer.Option("--provider", "-p")] = "hetzner",
):
"""List all active worker instances."""
from materia.workers import list_workers
workers = list_workers(provider)
if not workers:
typer.echo("No active workers")
return
typer.echo(f"{'NAME':<30} {'IP':<15} {'TYPE':<10} {'STATUS':<10}")
typer.echo("-" * 70)
for worker in workers:
typer.echo(f"{worker.name:<30} {worker.ip:<15} {worker.type:<10} {worker.status:<10}")
@worker_app.command("create")
def worker_create(
name: Annotated[str, typer.Argument(help="Worker name")],
server_type: Annotated[str, typer.Option("--type", "-t")] = "ccx22",
provider: Annotated[str, typer.Option("--provider", "-p")] = "hetzner",
location: Annotated[str | None, typer.Option("--location", "-l")] = None,
):
"""Create a new worker instance."""
from materia.workers import create_worker
typer.echo(f"Creating worker '{name}' ({server_type}) on {provider}...")
worker = create_worker(name, server_type, provider, location)
typer.echo(f"✓ Worker created: {worker.ip}")
@worker_app.command("destroy")
def worker_destroy(
name: Annotated[str, typer.Argument(help="Worker name")],
provider: Annotated[str, typer.Option("--provider", "-p")] = "hetzner",
force: Annotated[bool, typer.Option("--force", "-f")] = False,
):
"""Destroy a worker instance."""
from materia.workers import destroy_worker
if not force:
confirm = typer.confirm(f"Destroy worker '{name}'?")
if not confirm:
raise typer.Abort()
typer.echo(f"Destroying worker '{name}'...")
destroy_worker(name, provider)
typer.echo("✓ Worker destroyed")
pipeline_app = typer.Typer(help="Execute data pipelines")
app.add_typer(pipeline_app, name="pipeline")
@pipeline_app.command("run")
def pipeline_run(
name: Annotated[str, typer.Argument(help="Pipeline name (extract, transform)")],
):
"""Run a pipeline locally."""
from materia.pipelines import run_pipeline
typer.echo(f"Running pipeline '{name}'...")
result = run_pipeline(name)
if result.success:
typer.echo(result.output)
typer.echo("\n✓ Pipeline completed successfully")
else:
typer.echo(result.error, err=True)
raise typer.Exit(1)
@pipeline_app.command("list")
def pipeline_list():
"""List available pipelines."""
from materia.pipelines import PIPELINES
typer.echo("Available pipelines:")
for name, config in PIPELINES.items():
cmd = " ".join(config["command"])
typer.echo(f"{name:<15} (command: {cmd}, timeout: {config['timeout_seconds']}s)")
secrets_app = typer.Typer(help="Manage secrets via Pulumi ESC")
app.add_typer(secrets_app, name="secrets")
@secrets_app.command("list")
def secrets_list():
"""List available secrets (keys only)."""
from materia.secrets import list_secrets
secrets = list_secrets()
if not secrets:
typer.echo("No secrets configured")
return
typer.echo("Available secrets:")
for key in secrets:
typer.echo(f"{key}")
@secrets_app.command("get")
def secrets_get(
key: Annotated[str, typer.Argument(help="Secret key")],
):
"""Get a secret value."""
from materia.secrets import get_secret
value = get_secret(key)
if value is None:
typer.echo(f"Secret '{key}' not found", err=True)
raise typer.Exit(1)
typer.echo(value)
@secrets_app.command("test")
def secrets_test():
"""Test ESC connection and authentication."""
from materia.secrets import test_connection
typer.echo("Testing Pulumi ESC connection...")
if test_connection():
typer.echo("✓ ESC connection successful")
else:
typer.echo("✗ ESC connection failed", err=True)
typer.echo("\nMake sure you've run: esc login")
raise typer.Exit(1)
if __name__ == "__main__":
app()