PillarTool/pillar_tool/ptcli/main.py

import base64

import click

import requests
from urllib.parse import quote

from pillar_tool.schemas import HostCreateParams, HostgroupParams
from pillar_tool.util import config, load_config, Config
from pillar_tool.util.validation import split_and_validate_path, validate_fqdn

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("hostgroup")
def hostgroup():
    pass

@main.group("environment")
def environment():
    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}/host", headers=auth_header)
        response.raise_for_status()

        for h in response.json():
            click.echo(f"  - {h}")
    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:
        params = HostCreateParams(
            parent=parent
        )
        response = requests.post(f"{base_url}/host/{fqdn}", json=params.model_dump(), 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!")


@host.command("delete")
@click.argument("fqdn")
def host_delete(fqdn: str):
    if not validate_fqdn(fqdn):
        click.echo("Invalid FQDN")
        return

    if click.confirm(f"Are you sure you want to delete '{fqdn}'?"):
        click.echo("Deleting host...")

        try:
            response = requests.delete(f'{base_url}/host/{fqdn}', headers=auth_header)
            response.raise_for_status()
        except requests.exceptions.HTTPError as e:
            raise click.ClickException(f"Failed to delete host:\n{e}")


@hostgroup.command("list")
def hostgroup_list():
    click.echo("Listing known hostgroups...")
    try:
        response = requests.get(f'{base_url}/hostgroup', headers=auth_header)
        response.raise_for_status()

        click.echo("Hostgroups:")
        for hg in response.json():
            click.echo(f"  - {hg}")
    except requests.exceptions.HTTPError as e:
        raise click.ClickException(f"Failed to list hostgroups:\n{e}")


@hostgroup.command("show")
@click.argument("path")
def hostgroup_show(path: str):
    click.echo(f"Showing hostgroup '{path}'...")
    try:
        labels = split_and_validate_path(path)
        name = labels[-1]
        path = '/'.join(labels[:-1]) if len(labels) > 1 else None
        data = HostgroupParams(
            path=path
        )
        response = requests.get(f'{base_url}/hostgroup/{name}', headers=auth_header, params=data.model_dump())
        response.raise_for_status()
    except requests.exceptions.HTTPError as e:
        raise click.ClickException(f"Failed to show hostgroup:\n{e}")


@hostgroup.command("create")
@click.argument("path")
def hostgroup_create(path: str):
    click.echo(f"Creating hostgroup '{path}'...")
    try:
        labels = split_and_validate_path(path)
        path = "/".join(labels[:-1]) if len(labels) > 1 else ''
        name = labels[-1]
        data = HostgroupParams(
            path=path
        )
        response = requests.post(f'{base_url}/hostgroup/{name}', headers=auth_header, json=data.model_dump())
        response.raise_for_status()
    except requests.exceptions.HTTPError as e:
        raise click.ClickException(f"Failed to create hostgroup:\n{e}")


@hostgroup.command("delete")
@click.argument("path")
def hostgroup_delete(path: str):
    click.echo(f"Deleting hostgroup {path}...")
    try:
        labels = split_and_validate_path(path)
        name = labels[-1]
        prefix = "/".join(labels[:-1]) if len(labels) > 1 else None
        query_params = f"?path={prefix}" if prefix is not None else ''
        response = requests.delete(f'{base_url}/hostgroup/{name}{query_params}', headers=auth_header)
        response.raise_for_status()
    except requests.exceptions.HTTPError as e:
        raise click.ClickException(f"Failed to delete hostgroup:\n{e}")