""" Command-line application entry-point. Logic for the command-line spud tooling. """ import json import pathlib import click import httpx import uvicorn from spud.config import Configuration DEFAULT_CONFIGURATION_PATH = "~/.config/spud/config.json" @click.group() @click.option( "--config", type=str, default=None, help="Configuration path", ) @click.pass_context def cli(context, config): """CLI root""" context.ensure_object(dict) config = config if config is not None else DEFAULT_CONFIGURATION_PATH config_path = pathlib.Path(config).expanduser() context.obj["config_path"] = config_path if config_path.exists(): context.obj["config"] = Configuration.from_file(config_path) context.obj["api_client"] = httpx.Client( base_url=context.obj["config"].api_baseurl ) else: context.obj["config"] = None @click.command() @click.pass_context def init(context): """Generates a default configuration and writes it to disk.""" config_path = context.obj["config_path"] config_path.parent.mkdir(parents=True, exist_ok=True) if config_path.exists(): raise RuntimeError( f"File already exists ({str(config_path)}), cannot initialize." ) default_configuration = Configuration() with open(config_path, "w", encoding="utf8") as config_file: config_file.write(default_configuration.model_dump_json(indent=2)) @click.command() @click.pass_context def print_config(context): """Prints the current configuration.""" config_path = context.obj["config_path"] if not config_path.exists(): raise RuntimeError(f"Configuration file not found at {str(config_path)}.") with open(config_path, "r", encoding="utf8") as config_file: config_json = json.loads(config_file.read()) config = Configuration(**config_json) click.echo(config.model_dump_json(indent=2)) @click.command() @click.pass_context def status(context): """ Reports on the status of running services in a tree-like format. """ try: response = context.obj["api_client"].get("/status") data = response.json() except Exception as exc: # pylint: disable=broad-exception-caught click.secho(f"Failed to get services statuses: {str(exc)}", bold=True, fg="red") return if not data: click.secho("No running services to report on.", bold=True, fg="yellow") return for service in data: click.echo( click.style(f"{service['name']}", bold=True, fg="green") + f": {service['status']}" ) for container in service["containers"]: click.echo(f" + {container['name']} - {container['status']}") @click.command() @click.option( "--reload", type=bool, default=False, is_flag=True, help="Hot reloads on code updates.", ) def daemon(reload): """Starts the daemon process.""" uvicorn.run("spud.daemon:app", reload=reload) cli.add_command(init) cli.add_command(print_config) cli.add_command(status) cli.add_command(daemon) if __name__ == "__main__": cli(None, None)