#!/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())