Source code for freva_deployment.ui.deployment_tui.base

from __future__ import annotations
import os
from pathlib import Path
import logging

import curses
import npyscreen


logging.basicConfig(level=logging.DEBUG)
logger: logging.Logger = logging.getLogger("deploy-freva-tui")


[docs]class FileSelector(npyscreen.FileSelector): """FileSelector widget that allows for filtering file extensions.""" file_extentions: list[str] """List of allowed file extensions.""" value: str """The value of this file selector.""" def __init__( self, *args, file_extentions: str | list[str] = [".toml"], **kwargs ) -> None: if isinstance(file_extentions, str): self.file_extentions = [file_extentions] else: self.file_extentions = file_extentions super().__init__(*args, **kwargs) self.how_exited_handers[npyscreen.wgwidget.EXITED_ESCAPE] = self.exit
[docs] def exit(self) -> None: """Exit the dialogue.""" self.wCommand.value = "" self.value = "" self.exit_editing()
[docs] def update_grid(self) -> None: if self.value: self.value = os.path.expanduser(self.value) if not os.path.exists(self.value): self.value = os.getcwd() if os.path.isdir(self.value): working_dir = Path(self.value) else: working_dir = Path(self.value).parent self.wStatus1.value = working_dir file_list, dir_list = [], [] if working_dir.parent != working_dir: dir_list.append("..") try: for fn in working_dir.glob("*"): rel_path = str(fn.relative_to(working_dir)) if not rel_path.startswith("."): if fn.is_dir(): dir_list.append(str(fn) + os.sep) elif fn.suffix in self.file_extentions: file_list.append(str(fn)) except OSError: npyscreen.notify_wait( title="Error", message="Could not read specified directory." ) # DOES NOT CURRENTLY WORK - EXCEPT FOR THE WORKING DIRECTORY. REFACTOR. # sort Filelist file_list.sort(key=str.casefold) dir_list.sort(key=str.casefold) if self.sort_by_extension: file_list.sort(key=self.get_extension) self.wMain.set_grid_values_from_flat_list( dir_list + file_list, reset_cursor=False ) self.display()
[docs]def selectFile(starting_value: str = "", *args, **keywords): F = FileSelector(*args, **keywords) F.set_colors() F.wCommand.show_bold = True if starting_value: if not os.path.exists(os.path.abspath(os.path.expanduser(starting_value))): F.value = os.getcwd() else: F.value = starting_value F.wCommand.value = starting_value else: F.value = os.getcwd() F.update_grid() F.edit() return F.wCommand.value
[docs]class BaseForm(npyscreen.FormMultiPageWithMenus, npyscreen.FormWithMenus): """Base class for forms.""" _num: int = 0 input_fields: dict[str, tuple[npyscreen.TitleText, bool]] = {} """Dictionary of input fileds: the key of the dictionary represents the name of the key in the in config toml input files. Values represent a tuple of npysceen types that display the input information on this key to the user and a boolean indicating whether or not this variable is mandatory. """ certificates: list[str] = [] """The type of certificate files this step needs."""
[docs] def get_config(self, key) -> dict[str, str | bool | list[str]]: """Read the configuration for a step.""" try: cfg = self.parentApp.config[key].copy() if not isinstance(cfg, dict): cfg = {"config": {}} except (KeyError, AttributeError, TypeError): cfg = {"config": {}} cfg.setdefault("config", {}) for k, values in cfg["config"].items(): if values is None: cfg["config"][k] = "" return cfg["config"]
[docs] def get_host(self, key) -> str: """Read the host name(s) from the main windows config.""" try: host = self.parentApp.config[key]["hosts"] except (TypeError, KeyError): return "" if isinstance(host, str): host = [v.strip() for v in host.split(",") if v.strip()] return ",".join(host)
@property def num(self) -> str: """Calculate the number for enumerations of any input field.""" self._num += 1 return f"{self._num}. "
[docs] def check_config( self, notify: bool = True, ) -> dict[str, str | dict[str, str | list | int | bool | None]] | None: """Check if the from entries are valid.""" config = {} for key, (obj, mandatory) in self.input_fields.items(): try: value = obj.values[obj.value] except AttributeError: value = obj.value if isinstance(value, str): if not value and self.use.value and mandatory and notify: msg = f"MISSING ENTRY FOR {self.step}: {obj.name}" npyscreen.notify_confirm(msg, title="ERROR") return None elif not value and not mandatory: continue config[key] = value cfg = dict(hosts=config.pop("hosts")) cfg["config"] = config for key, value in cfg["config"].items(): if key in self.list_keys: cfg["config"][key] = value.split(",") return cfg
[docs] def draw_form(self) -> None: """Overload the draw_from method from, this is done to add menus.""" super().draw_form() menus = [ " " + self.__class__.MENU_KEY + ":Main Menu ", "^K:Prev. Tab ", "^L:Next Tab ", "^R:Run ", "^S:Save ", "^O:Load ", "^E:Exit ", ] y, x = self.display_menu_advert_at() for n, menu_advert in enumerate(menus): if isinstance(menu_advert, bytes): menu_advert = menu_advert.decode("utf-8", "replace") X = sum([len(menu) for menu in menus[:n]]) + x self.add_line( y, X, menu_advert, self.make_attributes_list(menu_advert, curses.A_NORMAL), self.columns - X - 1, )
[docs] def previews_form(self, *args, **keywords) -> None: return self.change_forms(*args, reverse=True, **keywords)
[docs] def change_forms(self, *args, reverse=False, **keywords) -> None: """Cycle between the deployment config forms.""" for step in self.parentApp._steps_lookup.keys(): if self.name.lower().startswith(step): name = step break elif self.name.lower().startswith("database"): name = "db" break keys = self.parentApp._steps_lookup.keys() steps = list(self.parentApp._steps_lookup.values()) if reverse: change_to = dict(zip(keys, [steps[-1]] + steps[:-1])) else: change_to = dict(zip(keys, steps[1:] + [steps[0]])) # raise ValueError(change_to, reverse, self.parentApp._steps_lookup) self.parentApp.current_form = name self.parentApp.change_form(change_to[name])
[docs] def run_deployment(self, *args) -> None: """Switch to the deployment setup form.""" self.parentApp.current_form = self.name self.parentApp.change_form("SETUP")
[docs] def create(self) -> None: """Setup the form.""" self.how_exited_handers[ npyscreen.wgwidget.EXITED_ESCAPE ] = self.parentApp.exit_application self.add_handlers({"^O": self.parentApp.load_dialog}) self.add_handlers({"^S": self.parentApp.save_dialog}) self.add_handlers({"^K": self.previews_form}) self.add_handlers({"^L": self.change_forms}) self.add_handlers({"^R": self.run_deployment}) self.add_handlers({"^E": self.parentApp.exit_application}) # The menus are created here. self.menu = self.add_menu(name="Main Menu", shortcut="^M") self.submenu = self.menu.addNewSubmenu("Deployment Menu", "^D") self._change_form = self.parentApp.change_form self.menu.addItemsFromList( [ ("Save Config", self.parentApp.save_dialog, "^S"), ("Load Config", self.parentApp.load_dialog, "^L"), ("Run Deployment", self.run_deployment, "^R"), ("Exit Application", self.parentApp.exit_application, "^E"), ] ) self.submenu.addItemsFromList( [ ("Core Deployment", self._change_form, "c", "c", ("MAIN",)), ("Web Deployment", self._change_form, "w", "w", ("SECOND",)), ("DB Deployment", self._change_form, "d", "d", ("THIRD",)), ("Solr Deployment", self._change_form, "s", "s", ("FOURTH",)), ("Run Deployment", self.run_deployment, "^R"), ] ) self.use = self.add( npyscreen.CheckBox, max_height=2, value=self.step in self.parentApp._steps, editable=True, name="Check to set up the {}".format(self.step), scroll_exit=True, ) self._add_widgets()