pkgbox

#!/usr/bin/env python
import json
from typing import Any, Dict, Optional

import click
from podman import PodmanClient


__VERSION__: str = '0.0.1'

DEFAULT_IMAGE: str = 'localhost/pkgbox/builder-base:latest'

PODMAN_URI = 'unix:///tmp/podman.sock'

client: PodmanClient  = PodmanClient(base_url=PODMAN_URI)


def jsonrpc_request(method: str, params: Dict[str, Any]) -> Dict[str, Any]:
    return {
        'id': None,
        'jsonrpc': '2.0',
        'method': method,
        'params': params
    }


def name_gen(image: str) -> str:
    parts = image.split('/')[-1].split(':')

    return f'pkgbox-buildbox-{parts[0]}-{parts[1]}'


@click.group
@click.option('--box', type=str)
@click.pass_context
def cli(ctx, box: str) -> None:
    ctx.ensure_object(dict)
    ctx.obj['box_name'] = box


@cli.command
def version(ctx) -> None:
    click.echo(f'v{__VERSION__}')


@cli.command
@click.option('--name', type=str, required=False)
@click.option('--image', type=str, default=DEFAULT_IMAGE)
@click.option('--create_only', is_flag=True, default=False)
def create(name: str, image: str, create_only: bool) -> None:
    click.echo('Testing API connection...')
    client.ping()

    click.echo(f'Checking if image "{image}" exists...')
    if not client.images.exists(image):
        click.echo(f'Image "{image}" does not exist, exiting...')
        return
    
    box_name = name if name is not None else name_gen(image)
    container = None
    click.echo(f'Checking if build box "{box_name}" exists...')
    if client.containers.exists(box_name):
        click.echo(f'Build box "{box_name}" already exists, skipping creation...')
        container = client.containers.get(box_name)
    else:
        click.echo(f'Creating build box "{box_name}"...')
        create_opts = {
            'name': box_name,
            'labels': {
                'pkgbox': 'true'
            },
            'environment': {
                'PKGBOX_NAME': box_name,
                'PKGBOX_IMAGE': image
            }
        }
        container = client.containers.create(image, ['sleep', 'infinity'], **create_opts)
        click.echo(f'Build box {box_name} created.')

    if create_only:
        click.echo('Will not start build box because "create-only" was set.')
    else:
        if container.status == 'running':
            click.echo(f'Build box "{box_name}" is already running.')
        else:
            click.echo(f'Starting build box {box_name}...')
            container.start()
            container.wait(condition=["running", "exited"])
            container.reload()
            click.echo(f'Build box {box_name} is running')


@cli.command
@click.argument('name')
@click.option('--confirm', prompt='All box data will be deleted, type YES to confirm')
def remove(name: str, confirm: str) -> None:
    if confirm != 'YES':
        return
    click.echo(f'Deleting build box {name} and its associated data...')

    if not client.containers.exists(name):
        click.echo(f'Build box does not exist: "{name}", exiting...')
        
        return
 
    remove_opts = {
        'force': True
    }
    client.containers.remove(name, **remove_opts)

    click.echo(f'Build box {name} deleted.')


@cli.command
def list():
    list_opts = {
        'labels': {
            'pkgbox': 'true'
        }
    }
    containers = client.containers.list(**list_opts)

    if len(containers) == 0:
        return

    for container in containers:
        click.echo(container.name)


@cli.command
@click.argument('name')
def status(name: str):
    if not client.containers.exists(name):
        click.echo(f'Box not found: {name}')
        return

    container = client.containers.get(name)

    click.echo(container.status)


@cli.command
@click.pass_context
@click.option('--repository', type=str, required=False)
def build(ctx, repository: Optional[str]):
    box_name  = ctx.obj.get('box_name')
    if not client.containers.exists(box_name):
        click.echo(f'Box not found: {box_name}')
        return

    container = client.containers.get(name)

    cmd_params = {
        'repository': repository
    }
    cmd = [
        '/opt/pkgbox/pkgbox-rpc',
        json.dumps(jsonrpc_request('build', cmd_params))
    ]
    code, out = container.exec_run(cmd)

    if code != 0:
        click.echo(f'Error: {out}')
        return    

    data = json.loads(out.decode().replace('\n', ''))
    click.echo(data)


if __name__ == '__main__':
    cli()