speech-collect/tasks.py

import os

from invoke import task

SPEECH_COLLECT_RESPOSITORY = os.environ.get("SPEECH_COLLECT_RESPOSITORY", "registry.runcible.io/speech-collect")
SPEECH_COLLECT_HELM_CHART_DIR = "./charts/speech-collect"
SPEECH_COLLECT_HELM_REPO = os.environ.get(
    "SPEECH_COLLECT_HELM_REPO", "https://git.runcible.io/api/packages/androiddrew/helm/api/charts"
)


@task
def serve(c):
    """Serves the speech-collect application locally."""
    c.run("LITESTAR_APP=speech_collect.app:app litestar run --port 8888 --host 0.0.0.0 --reload")


@task
def update_deps(c):
    """Updates depenencies"""
    c.run("pip-compile requirements.in", pty=True)
    c.run("pip-compile dev-requirements.in", pty=True)


@task
def sync_deps(c):
    """Syncs local dependencies"""
    c.run("pip-sync requirements.txt dev-requirements.txt")


@task
def build_image(c, dev=True, registry_user=None, registry_token=None, push=False, login=False):
    """Builds the speech-collect container image."""
    context_dir = c.run("pwd", hide=True).stdout.strip()
    commit_sha = c.run("git rev-parse --short HEAD", hide=True).stdout.strip()
    image_name = f"{SPEECH_COLLECT_RESPOSITORY}:{commit_sha}{'-dev' if dev else ''}"
    c.run(f"docker build --build-arg='COMMIT_SHA={commit_sha}' -t {image_name} {context_dir}")
    if login:
        if registry_user is None or registry_token is None:
            raise ValueError("--registry_user and --registry_token must be provided if using --login parameter")
        c.run(f"docker login -u {registry_user} -p {registry_token}")
    if push:
        c.run(f"docker push {image_name}")


@task
def delint(c):
    """Applies automated linters to project"""
    c.run("isort ./speech_collect ./tests ./tasks.py", pty=True)
    c.run("black ./speech_collect ./tests ./tasks.py", pty=True)


@task
def template_chart(c, values=None):
    """Templates the speech-collect Helm chart with an optional custom values.yaml"""
    cmd = f"helm template {SPEECH_COLLECT_HELM_CHART_DIR}"
    if values:
        cmd += f" --values {values}"
    c.run(cmd)


@task
def rebuild_chart(c):
    """Rebuilds the speech-collect helm chart"""
    c.run(f"ls {SPEECH_COLLECT_HELM_CHART_DIR}")
    c.run(f"rm -rf {SPEECH_COLLECT_HELM_CHART_DIR}/Chart.lock {SPEECH_COLLECT_HELM_CHART_DIR}/charts")
    # c.run(f"helm dependency update --respository-config .charts/repositories.yaml {SPEECH_COLLECT_HELM_CHART_DIR}")


@task(pre=[rebuild_chart])
def package_chart(c, pre_release=None, destination="./build"):
    """Packages the speech-collect helm chart."""
    # Helm requires a valid semantic version.
    # See https://semver.org
    chart_version = c.run(
        f"helm show chart {SPEECH_COLLECT_HELM_CHART_DIR} | grep version | tail -1 | cut -d ':' -f2 | xargs"
    ).stdout.strip()
    if pre_release is not None:
        chart_version = chart_version + str(pre_release)
    c.run(f"helm package {SPEECH_COLLECT_HELM_CHART_DIR} --version {chart_version} --destination {destination}")


@task(pre=[package_chart])
def publish_chart(c, repo_user=None, repo_token=None, build_dir="./build"):
    """Pushes a helm chart to a repository"""
    chart_name = c.run(
        f"helm show chart {SPEECH_COLLECT_HELM_CHART_DIR} | grep name | tail -1 | cut -d ':' -f2 | xargs"
    ).stdout.strip()
    chart_version = c.run(
        f"helm show chart {SPEECH_COLLECT_HELM_CHART_DIR} | grep version | tail -1 | cut -d ':' -f2 | xargs"
    ).stdout.strip()
    target_chart = f"{chart_name}-{chart_version}.tgz"
    cmd = f"curl --user {repo_user}:{repo_token} -X POST --upload-file {build_dir}/{target_chart} {SPEECH_COLLECT_HELM_REPO}"
    c.run(cmd)