From 5dc7ef76e883bde38b28067b1f4d107c17935b82 Mon Sep 17 00:00:00 2001 From: Micke Nordin Date: Fri, 17 Jan 2025 11:44:01 +0100 Subject: [PATCH] Begin refactor --- knotctl/__init__.py | 93 ++++++++++---------------------------- knotctl/config/__init__.py | 61 +++++++++++++++++++++++++ knotctl/utils/__init__.py | 35 ++++++++++++++ 3 files changed, 119 insertions(+), 70 deletions(-) create mode 100755 knotctl/config/__init__.py create mode 100644 knotctl/utils/__init__.py diff --git a/knotctl/__init__.py b/knotctl/__init__.py index a768fd2..4a6bae3 100755 --- a/knotctl/__init__.py +++ b/knotctl/__init__.py @@ -2,12 +2,9 @@ import argparse import getpass -import json import os import sys import urllib.parse -from os import environ, mkdir -from os.path import isdir, isfile, join from typing import Union from urllib.parse import urlparse @@ -19,30 +16,29 @@ import yaml from requests.models import HTTPBasicAuth from simplejson.errors import JSONDecodeError as SimplejsonJSONDecodeError +from .config import Config +from .utils import error, output + try: from requests.exceptions import JSONDecodeError as RequestsJSONDecodeError except ImportError: from requests.exceptions import InvalidJSONError as RequestsJSONDecodeError -# Helper functions -def error(description: str, error: str) -> list[dict]: - response = [] - reply = {} - # https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/406 - reply["Code"] = 406 - reply["Description"] = description - reply["Error"] = error - response.append(reply) - return response +class Knotctl: + + def __init__(self): + self.conf = Config() + self.config = self.get_config() + self.config_filename = self.conf.config_filename + def get_config(self): + config = self.conf.get_config() + if not config: + print("You need to configure knotctl before proceeding") + run_config() -def get_config(config_filename: str): - if not isfile(config_filename): - print("You need to configure knotctl before proceeding") - run_config(config_filename) - with open(config_filename, "r") as fh: - return yaml.safe_load(fh.read()) + return config def get_openstack_addresses(cloud: str, name: str): @@ -60,30 +56,6 @@ def get_openstack_addresses(cloud: str, name: str): return openstack_addresses -def nested_out(input, tabs="") -> str: - string = "" - if isinstance(input, str) or isinstance(input, int): - string += "{}\n".format(input) - elif isinstance(input, dict): - for key, value in input.items(): - string += "{}{}: {}".format(tabs, key, - nested_out(value, tabs + " ")) - elif isinstance(input, list): - for entry in input: - string += "{}\n{}".format(tabs, nested_out(entry, tabs + " ")) - return string - - -def output(response: list[dict], jsonout: bool = False): - try: - if jsonout: - print(json.dumps(response)) - else: - print(nested_out(response)) - except BrokenPipeError: - pass - - # Define the runner for each command def run_add(url: str, jsonout: bool, headers: dict): parsed = split_url(url) @@ -152,7 +124,6 @@ def run_complete(shell: Union[None, str]): def run_config( - config_filename: str, context: Union[None, str] = None, baseurl: Union[None, str] = None, list_config: bool = False, @@ -160,29 +131,18 @@ def run_config( password: Union[None, str] = None, current: Union[None, str] = None, ): + conf = Config() if current: - if os.path.islink(config_filename): - actual_path = os.readlink(config_filename) - print(actual_path.split("-")[-1]) - else: - print("none") + print(conf.get_current()) return config = {"baseurl": baseurl, "username": username, "password": password} needed = [] if context: - symlink = f"{config_filename}-{context}" - found = os.path.isfile(symlink) - if os.path.islink(config_filename): - os.remove(config_filename) - elif os.path.isfile(config_filename): - os.rename(config_filename, symlink) - os.symlink(symlink, config_filename) - config_filename = symlink + found = conf.set_context(context) if found: return if list_config: - config_data = get_config(config_filename) - config_data.pop("password", None) + config_data = conf.get_config_data() output(config_data) return if not baseurl: @@ -206,8 +166,7 @@ def run_config( output(error("Can not configure without password", "No password")) sys.exit(1) - with open(config_filename, "w") as fh: - fh.write(yaml.dump(config)) + conf.set_config(config) def run_delete(url: str, jsonout: bool, headers: dict): @@ -280,7 +239,7 @@ def run_openstack_sync(cloud: str, name: str, zone: str, headers: dict, elif address.version == 6: rtype = "AAAA" curripv6 = True - if rtype and recor.type == rtype: + if rtype and record.type == rtype: if record.data == address.addr: continue else: @@ -665,16 +624,10 @@ def main() -> int: run_complete(args.shell) return 0 - # Make sure we have config - config_basepath = join(environ["HOME"], ".knot") - config_filename = join(config_basepath, "config") - - if not isdir(config_basepath): - mkdir(config_basepath) + knotctl = Knotctl() if args.command == "config": run_config( - config_filename, args.context, args.baseurl, args.list_config, @@ -684,7 +637,7 @@ def main() -> int: ) return 0 - config = get_config(config_filename) + config = knotctl.get_config() baseurl = config["baseurl"] token = get_token(config) if token == "": diff --git a/knotctl/config/__init__.py b/knotctl/config/__init__.py new file mode 100755 index 0000000..7c2faff --- /dev/null +++ b/knotctl/config/__init__.py @@ -0,0 +1,61 @@ +#!/usr/bin/env python3 +import os +from os import mkdir +from os.path import isdir, isfile, join +from typing import Union + +import yaml + + +class Config: + + def __init__(self): + # Make sure we have config + self.config_basepath = join(os.environ["HOME"], ".knot") + self.config_filename = join(self.config_basepath, "config") + if not isdir(self.config_basepath): + mkdir(self.config_basepath) + + def get_config(self) -> Union[None, dict]: + if not isfile(self.config_filename): + return None + with open(self.config_filename, "r") as fh: + return yaml.safe_load(fh.read()) + + def get_config_data(self) -> dict: + config_data = self.get_config() + config_data.pop("password", None) + return config_data + + def get_current(self) -> str: + if os.path.islink(self.config_filename): + actual_path = os.readlink(self.config_filename) + return actual_path.split("-")[-1] + else: + return "none" + + def set_context(self, context) -> bool: + symlink = f"{self.config_filename}-{context}" + found = os.path.isfile(symlink) + if os.path.islink(self.config_filename): + os.remove(self.config_filename) + elif os.path.isfile(self.config_filename): + os.rename(self.config_filename, symlink) + os.symlink(symlink, self.config_filename) + self.config_filename = symlink + return found + + def set_config( + self, + baseurl: str, + username: str, + password: str, + ): + config = { + "baseurl": baseurl, + "username": username, + "password": password + } + + with open(self.config_filename, "w") as fh: + fh.write(yaml.dump(config)) diff --git a/knotctl/utils/__init__.py b/knotctl/utils/__init__.py new file mode 100644 index 0000000..501f36f --- /dev/null +++ b/knotctl/utils/__init__.py @@ -0,0 +1,35 @@ +import json + + +def error(description: str, error: str): + response = [] + reply = {} + # https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/406 + reply["Code"] = 406 + reply["Description"] = description + reply["Error"] = error + response.append(reply) + return response + + +def nested_out(input, tabs="") -> str: + string = "" + if isinstance(input, str) or isinstance(input, int): + string += f"{input}\n" + elif isinstance(input, dict): + for key, value in input.items(): + string += f"{tabs}{key}: {nested_out(value, tabs + " ")}" + elif isinstance(input, list): + for entry in input: + string += f"{tabs}\n{nested_out(entry, tabs + ' ')}" + return string + + +def output(response: list[dict], jsonout: bool = False): + try: + if jsonout: + print(json.dumps(response)) + else: + print(nested_out(response)) + except BrokenPipeError: + pass