infra-as-code/service.py

#!/usr/bin/env python3
from __future__ import annotations
from pathlib import Path
from os import path
import subprocess as sub
import re
from typing import Any, Callable
from enum import Enum

P_BANNER = """
██╗    ██╗███████╗██╗     ██╗  ██╗ ██████╗ ███╗   ███╗
██║    ██║██╔════╝██║     ██║ ██╔╝██╔═══██╗████╗ ████║
██║ █╗ ██║█████╗  ██║     █████╔╝ ██║   ██║██╔████╔██║
██║███╗██║██╔══╝  ██║     ██╔═██╗ ██║   ██║██║╚██╔╝██║
╚███╔███╔╝███████╗███████╗██║  ██╗╚██████╔╝██║ ╚═╝ ██║
 ╚══╝╚══╝ ╚══════╝╚══════╝╚═╝  ╚═╝ ╚═════╝ ╚═╝     ╚═╝
"""

P_OPTIONS = """
Opties:
0) Afsluiten
1) Maak/Wijzig omgeving(en)
2) Vernietig omgeving(en)
3) Recover omgeving(en)
4) List omgeving(en)
"""
class PortalOptions(Enum):
    EXIT = "0"
    CREATE = "1"
    DESTROY = "2"
    RECOVER = "3"
    LIST = "4"


RE_TEXT = re.compile(r"^\w+$")
RE_NUM = re.compile(r"^\d+$")
RE_ANY = re.compile(r".+")


class Prompter:

    def __init__(self, ps1: str = "") -> None:
        self.ps1 = ps1

    def _build_prompt(self, prompt: str) -> str:
        return self.ps1 + prompt

    def take_input(self,
                   prompt: str,
                   pattern: re.Pattern[Any],
                   default: str = "") -> str:
        prompt = self._build_prompt(prompt)
        return Prompter.input(prompt, lambda x: bool(pattern.match(x)),
                              default)

    def take_choice(self, prompt: str, options: list[str]) -> str:
        prompt = self._build_prompt(prompt)
        return Prompter.input(prompt, lambda x: x in options)

    def take_confirmation(self, prompt: str, default: str = "y") -> bool:
        prompt = self._build_prompt(prompt)
        i = Prompter.input(prompt, lambda x: x.lower() in ["y", "n"], default=default)   
        return i == "y"

    @staticmethod
    def input(prompt: str,
              validate: Callable[[str], bool],
              default: str = "") -> str:
        while True:
            i = input(prompt)
            if i == "":
                i = default
            if validate(i):
                return i

def get_env_list(customer: str) -> list[str]:
    """Fetches and parses a list of a customer's environments

    Args:
        customer (str): The customer to fetch environments from

    Returns:
        list[str]: The parsed environments
    """
    return sub.check_output(["python3", "cli.py", "list",
                             customer]).decode().strip("\n").split(" ")


def main() -> int:
    print(P_BANNER)
    customer_name = Prompter.input("Wat is uw naam: ",
                                   lambda x: bool(RE_TEXT.match(x)))
    """
    Ensures the necessary customer setup.
    
    NOTE: The customer and the `envs` directory inside of it never get removed.
    """
    envs_dir = path.join("customers", customer_name, "envs")
    Path(envs_dir).mkdir(parents=True, exist_ok=True)

    p = Prompter(ps1=f"{customer_name} > ")
    while True:
        print(P_OPTIONS)
        choice = p.take_input("Keuze: ", RE_ANY)
        print(end="\n")

        if choice == PortalOptions.EXIT.value:
            """
            Used to break out of the loop and exit the portal.
            """
            break

        if choice == PortalOptions.CREATE.value:
            """
            Used to either update or create an environment.
            """
            envs = get_env_list(customer_name)
            print(
                f"NOTE: Kies een albestaande omgevingen {envs} of iets nieuws..."
            )
            fmt = "Omgevingsnaam: "
            env_name = p.take_input(fmt, RE_TEXT)

            # Loop until customer has confirmed their desired combination of machines
            while True:
                templates = ["production", "acceptance", "test", "custom"]
                fmt = f"Het type omgeving {templates}: "
                template_name = p.take_choice(fmt, templates)
                if template_name == "custom":
                    """
                    Asks for all machine's individually
                    """
                    fmt = "Aantal nginx webservers (default=1): "
                    amnt_nginx_web = p.take_input(fmt, RE_NUM, default="1")

                    fmt = "Aantal nginx loadbalancers (default=1): "
                    amnt_nginx_lb = p.take_input(fmt, RE_NUM, default="1")

                    fmt = "Aantal postgres instances (default=1): "
                    amnt_psql = p.take_input(fmt, RE_NUM, default="1") 
                elif template_name == "production":
                    amnt_nginx_web = "2"
                    amnt_nginx_lb = "1"
                    amnt_psql = "1"
                    
                elif template_name == "acceptance":
                    amnt_nginx_web = "1"
                    amnt_nginx_lb = "0"
                    amnt_psql = "1"
                # elif template_name == "test":
                else:
                    amnt_nginx_web = "1"
                    amnt_nginx_lb = "0"
                    amnt_psql = "0"
                
                # TODO: migrate different machine types to a dict model
                print(end="\n")
                print("Deze omgeving bevat:")
                if amnt_nginx_web > "0":
                    print(f" - {amnt_nginx_web} Nginx webserver(s)")
                
                if amnt_nginx_lb > "0":
                    print(f" - {amnt_nginx_lb} Nginx loadbalancer(s)")

                if amnt_psql > "0":
                    print(f" - {amnt_psql} Postgres instance(s)")
                print(end="\n")

                if p.take_confirmation("Bevestig (Y/n): "):
                    break

            """
            Define the format for templating ip-addresses.`{}` is required 
            and will be subbed at runtime with the correct octet.
            """
            print(end="\n")
            fmt = "NOTE: `{}` will be replaced with the correct octet at runtime"
            print(fmt)
            fmt = "Format of ip (default=192.168.56.{}): "
            ip_format = p.take_input(fmt, RE_ANY, default="192.168.56.{}")

            fmt = "Number to start formatting the IP from (default=10): "
            ip_int = p.take_input(fmt, RE_NUM, default="10")

            sub.call([
                "python3", "cli.py", "create", customer_name, env_name,
                "--num-nginx-web", amnt_nginx_web, "--num-postgres", amnt_psql,
                "--num-nginx-lb", amnt_nginx_lb, "--ip-format", ip_format,
                "--ip-int", ip_int
            ])
            print(f"Omgeving `{env_name}` successvol gemaakt.")

        if choice == PortalOptions.DESTROY.value:
            """
            Deletes all traces of the chosen environment.
            """
            envs = get_env_list(customer_name)
            fmt = f"Omgevingsnaam {envs}: "
            env_name = p.take_choice(fmt, envs)
            sub.call(["python3", "cli.py", "delete", customer_name, env_name])

        if choice == PortalOptions.RECOVER.value:
            """
            Allows the customer to "force" their environment into the desired `up-state`. 
            This `up-state` is a representation of the Vagrantfile, Ansible inventory and 
            .ssh directory of that specific environment. The .ssh directory is only 
            necessary in case of an existing storage-volume. Otherwise a new one 
            keypair can be manually created or created using option 1.
            """
            envs = get_env_list(customer_name)
            fmt = f"Omgevingsnaam {envs}: "
            env_name = p.take_choice(fmt, envs)
            sub.call(["python3", "cli.py", "recover", customer_name, env_name])

        if choice == PortalOptions.LIST.value:
            """
            This branch displays the customer's existing environments
            """
            print("De volgende omgeving(en) zijn aanwezig: ")
            for env in get_env_list(customer_name):
                print(f" - {env}")

    return 0


if __name__ == "__main__":
    raise SystemExit(main())