import base64 import click import requests from click import Context from pillar_tool.util import config, load_config, Config cfg: Config | None = None base_url: str | None = None auth_header: dict[str, str] | None = None @click.group("command") def main(): global cfg, base_url, auth_header # load the configuration and store it load_config() cfg = config() base_url = f"{cfg.ptcli.scheme}://{cfg.ptcli.host}:{cfg.ptcli.port}" auth_header = { 'Authorization': f"Basic {base64.b64encode(f"{cfg.ptcli.user}:{cfg.ptcli.password}".encode('utf-8')).decode('ascii')}" } # health check of the api try: response = requests.get(f"{base_url}/health") response.raise_for_status() except requests.exceptions.HTTPError as e: raise click.ClickException(f"API seems to be unhealthy:\n{e}") except requests.exceptions.ConnectionError as e: raise click.ClickException("Unable to connect to PillarTool API") @main.group("pillar") def pillar(): pass @main.group("host") def host(): pass @main.group("query") def query(): pass @pillar.command("get") def pillar_get(): print("pillar_get") @pillar.command("list") def pillar_list(): print("pillar_list") @host.command("list") def host_list(): click.echo("Listing known hosts...") try: response = requests.get(f"{base_url}/hosts", headers=auth_header) response.raise_for_status() print(response.json()) except requests.exceptions.HTTPError as e: raise click.ClickException(f"Failed to list hosts:\n{e}") @host.command("create") @click.argument("fqdn") @click.argument("parent", default=None) def host_create(fqdn: str, parent: str | None): click.echo("Creating host...") try: response = requests.post(f"{base_url}/host/{fqdn}", json={'parent': parent}, headers=auth_header) response.raise_for_status() except requests.exceptions.HTTPError as e: raise click.ClickException(f"Failed to create host:\n{e}") click.echo(f"Host '{fqdn}' created!")