magic-app/spikes/first.py

"""The all singing all dancing auto-generation from compose.yml attempt.

Largely unfinished, just a stab in the dark.

"""

from json import loads
from os.path import exists
from pathlib import Path
from re import findall
from shlex import split
from subprocess import run

from flask import Flask, render_template, send_from_directory
from flask_wtf import FlaskForm
from ruamel.yaml import YAML
from wtforms import StringField
from wtforms.validators import DataRequired

app = Flask(__name__)

# Note(decentral1se): load from env vars at some point
app.secret_key = b'_5#y2L"F4Q8z\n\xec]/'

yaml = YAML()

# Note(decentral1se): paths needs to change dependening
# on deployment mode ./data in dev, /data in production
DATA_DIR = Path("./data")

STATIC_DIR = Path(".").absolute() / "static"


class InstallAppInputForm(FlaskForm):
    """Dynamically generated input form for app deploy."""

    pass


def get_apps_json():
    """Retrieve the apps listing."""
    global STATIC_DIR

    try:
        apps_path = STATIC_DIR / "apps.json"
        with open(apps_path, "r") as handle:
            return loads(handle.read())
    except Exception as exception:
        app.logger.error(f"Failed to load apps.json, saw {exception}")


def clone_app_template(app_name):
    """Git clone an app template repository."""
    global DATA_DIR

    try:
        apps_json = get_apps_json()
        app_repo_url = apps_json[app_name]
    except KeyError:
        app.logger.error(f"No repository for {app_name} found")
        # Note(decentral1se): should catch this or make more specific error
        raise RuntimeError()

    try:
        clone_path = DATA_DIR / app_name
        command = f"git clone {app_repo_url} {clone_path}"
        run(split(command))
    except Exception as exception:
        app.logger.error(f"Failed to run {command}, saw {exception}")
        # Note(decentral1se): should catch this or make more specific error
        raise RuntimeError()


def get_env_vars(path):
    """Get required env vars from a compose.yml."""
    try:
        compose_path = Path(path) / "compose.yml"
        with open(compose_path, "r") as handle:
            compose_yml = handle.read()
    except Exception as exception:
        app.logger.error(f"Failed to load {compose_path}, saw {exception}")
        # Note(decentral1se): should catch this or make more specific error
        raise RuntimeError()

    # Note(decentral1se): matches all env vars (e.g ${FOO} and returns FOO)
    return findall(r"\${(.*?)}", compose_yml)


def get_secrets(path):
    """Get required secrets from a compose.yml."""
    try:
        compose_path = Path(path) / "compose.yml"
        with open(compose_path, "r") as handle:
            compose_yml = yaml.load(handle.read())
    except Exception as exception:
        app.logger.error(f"Failed to load {compose_path}, saw {exception}")
        # Note(decentral1se): should catch this or make more specific error
        raise RuntimeError()

    # Note(decentral1se): retrieves top level names like "db_passwd" not
    # "gitea_db_passwd_v1" and I am unsure at this point which is the best to
    # bring in but at least the plumbing is there
    return [secret for secret in compose_yml["secrets"].keys()]


def get_compose_inputs(path):
    """Retrieve a list of all required inputs for stack deploy."""
    return {
        "env_vars": get_env_vars(path),
        "secrets": get_secrets(path),
    }


@app.route("/")
def home():
    """Home page."""
    apps = [app_name for app_name in get_apps_json()]
    return render_template("first/index.html", apps=apps)


@app.route("/apps")
def apps_list():
    """The JSON list of all apps."""
    return send_from_directory("static", "apps.json")


@app.route("/<app_name>")
def install_app(app_name):
    """Install an app"""
    clone_app_template(app_name)

    inputs = get_compose_inputs(DATA_DIR / app_name)

    for env_var in inputs["env_vars"]:
        setattr(InstallAppInputForm, env_var, StringField())

    for secret in inputs["secrets"]:
        setattr(InstallAppInputForm, secret, StringField())

    form = InstallAppInputForm()

    return render_template("first/install.html", app_name=app_name, form=form)


@app.route("/deploy/<app_name>")
def deploy_app(app_name):
    return "TODO"


if __name__ == "__main__":
    app.run(debug=True, host="0.0.0.0")