From 0b37e006b1b73aa632ff55e5edcdbd54ec919051 Mon Sep 17 00:00:00 2001 From: Frank Brehm Date: Tue, 20 Mar 2018 15:07:26 +0100 Subject: [PATCH] Adding bin/create-vmware-template and some modules --- bin/create-vmware-template | 44 +++ lib/cr_vmware_tpl/app.py | 741 +++++++++++++++++++++++++++++++++++ lib/cr_vmware_tpl/colored.py | 215 ++++++++++ lib/cr_vmware_tpl/errors.py | 153 ++++++++ lib/cr_vmware_tpl/obj.py | 328 ++++++++++++++++ 5 files changed, 1481 insertions(+) create mode 100755 bin/create-vmware-template create mode 100644 lib/cr_vmware_tpl/app.py create mode 100644 lib/cr_vmware_tpl/colored.py create mode 100644 lib/cr_vmware_tpl/errors.py create mode 100644 lib/cr_vmware_tpl/obj.py diff --git a/bin/create-vmware-template b/bin/create-vmware-template new file mode 100755 index 0000000..2a08f17 --- /dev/null +++ b/bin/create-vmware-template @@ -0,0 +1,44 @@ +#!/usr/bin/env python3 + +# Standard modules +import sys +import os +import logging +import locale + +# own modules: +cur_dir = os.getcwd() +base_dir = cur_dir + +if sys.argv[0] != '' and sys.argv[0] != '-c': + bin_dir = os.path.dirname(os.path.realpath(sys.argv[0])) +else: + bin_dir = os.path.dirname(os.path.realpath(__file__)) +base_dir = os.path.abspath(os.path.join(bin_dir, '..')) +lib_dir = os.path.join(base_dir, 'lib') +module_dir = os.path.join(lib_dir, 'cr_vmware_tpl') +if os.path.exists(module_dir): + sys.path.insert(0, base_dir) + +from cr_vmware_tpl.app import CrTplApplication + +log = logging.getLogger(__name__) + +__author__ = 'Frank Brehm ' +__copyright__ = '(C) 2018 by Frank Brehm, Pixelpark GmbH, Berlin' + +appname = os.path.basename(sys.argv[0]) + +locale.setlocale(locale.LC_ALL, '') + +app = CrTplApplication(appname=appname) +app.initialized = True + +if app.verbose > 2: + print("{c}-Object:\n{a}".format(c=app.__class__.__name__, a=app)) + +app() + +sys.exit(0) + +# vim: tabstop=4 expandtab shiftwidth=4 softtabstop=4 diff --git a/lib/cr_vmware_tpl/app.py b/lib/cr_vmware_tpl/app.py new file mode 100644 index 0000000..792c14d --- /dev/null +++ b/lib/cr_vmware_tpl/app.py @@ -0,0 +1,741 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +""" +@author: Frank Brehm +@contact: frank.brehm@pixelpark.com +@copyright: © 2018 by Frank Brehm, Berlin +@summary: The module for the application object. +""" +from __future__ import absolute_import + +# Standard modules +import sys +import os +import logging +import re +import traceback + +# Third party modules +import argparse + +# Own modules +from .errors import FunctionNotImplementedError, PpAppError + +from .common import terminal_can_colors +from .common import caller_search_path + +from .colored import ColoredFormatter, colorstr + +from .obj import PpBaseObject + +__version__ = '0.1.1' +LOG = logging.getLogger(__name__) + + +# ============================================================================= +class CrTplApplication(PpBaseObject): + """ + Class for the application objects. + """ + + re_prefix = re.compile(r'^[a-z0-9][a-z0-9_]*$', re.IGNORECASE) + re_anum = re.compile(r'[^A-Z0-9_]+', re.IGNORECASE) + + # ------------------------------------------------------------------------- + def __init__( + self, appname=None, verbose=0, version=__version__, base_dir=None, + initialized=False, usage=None, description=None, + argparse_epilog=None, argparse_prefix_chars='-', env_prefix=None): + + self.arg_parser = None + """ + @ivar: argparser object to parse commandline parameters + @type: argparse.ArgumentParser + """ + + self.args = None + """ + @ivar: an object containing all commandline parameters + after parsing them + @type: Namespace + """ + + self._exit_value = 0 + """ + @ivar: return value of the application for exiting with sys.exit(). + @type: int + """ + + self._usage = usage + """ + @ivar: usage text used on argparse + @type: str + """ + + self._description = description + """ + @ivar: a short text describing the application + @type: str + """ + + self._argparse_epilog = argparse_epilog + """ + @ivar: an epilog displayed at the end of the argparse help screen + @type: str + """ + + self._argparse_prefix_chars = argparse_prefix_chars + """ + @ivar: The set of characters that prefix optional arguments. + @type: str + """ + + self._terminal_has_colors = False + """ + @ivar: flag, that the current terminal understands color ANSI codes + @type: bool + """ + + self._quiet = False + + self.env = {} + """ + @ivar: a dictionary with all application specifiv environment variables, + they will detected by the env_prefix property of this object, + and their names will transformed before saving their values in + self.env by removing the env_prefix from the variable name. + @type: dict + """ + + self._env_prefix = None + """ + @ivar: a prefix for environment variables to detect them and to assign + their transformed names and their values in self.env + @type: str + """ + + super(CrTplApplication, self).__init__( + appname=appname, + verbose=verbose, + version=version, + base_dir=base_dir, + initialized=False, + ) + + if env_prefix: + ep = str(env_prefix).strip() + if not ep: + msg = "Invalid env_prefix {!r} given - it may not be empty.".format(env_prefix) + raise PpAppError(msg) + match = self.re_prefix.search(ep) + if not match: + msg = ( + "Invalid characters found in env_prefix {!r}, only " + "alphanumeric characters and digits and underscore " + "(this not as the first character) are allowed.").format(env_prefix) + raise PpAppError(msg) + self._env_prefix = ep + else: + ep = self.appname.upper() + '_' + self._env_prefix = self.re_anum.sub('_', ep) + + self._init_arg_parser() + self._perform_arg_parser() + + self._init_env() + self._perform_env() + + # ----------------------------------------------------------- + @property + def exit_value(self): + """The return value of the application for exiting with sys.exit().""" + return self._exit_value + + @exit_value.setter + def exit_value(self, value): + v = int(value) + if v >= 0: + self._exit_value = v + else: + LOG.warn("Wrong exit_value {!r}, must be >= 0".format(value)) + + # ----------------------------------------------------------- + @property + def exitvalue(self): + """The return value of the application for exiting with sys.exit().""" + return self._exit_value + + @exitvalue.setter + def exitvalue(self, value): + self.exit_value = value + + # ----------------------------------------------------------- + @property + def usage(self): + """The usage text used on argparse.""" + return self._usage + + # ----------------------------------------------------------- + @property + def description(self): + """A short text describing the application.""" + return self._description + + # ----------------------------------------------------------- + @property + def argparse_epilog(self): + """An epilog displayed at the end of the argparse help screen.""" + return self._argparse_epilog + + # ----------------------------------------------------------- + @property + def argparse_prefix_chars(self): + """The set of characters that prefix optional arguments.""" + return self._argparse_prefix_chars + + # ----------------------------------------------------------- + @property + def terminal_has_colors(self): + """A flag, that the current terminal understands color ANSI codes.""" + return self._terminal_has_colors + + # ----------------------------------------------------------- + @property + def env_prefix(self): + """A prefix for environment variables to detect them.""" + return self._env_prefix + + # ----------------------------------------------------------- + @property + def usage_term(self): + """The localized version of 'usage: '""" + return 'Usage: ' + + # ----------------------------------------------------------- + @property + def usage_term_len(self): + """The length of the localized version of 'usage: '""" + return len(self.usage_term) + + # ----------------------------------------------------------- + @property + def quiet(self): + """Quiet execution of the application, + only warnings and errors are emitted.""" + return self._quiet + + @quiet.setter + def quiet(self, value): + self._quiet = bool(value) + + # ------------------------------------------------------------------------- + def exit(self, retval=-1, msg=None, trace=False): + """ + Universal method to call sys.exit(). If fake_exit is set, a + FakeExitError exception is raised instead (useful for unittests.) + + @param retval: the return value to give back to theoperating system + @type retval: int + @param msg: a last message, which should be emitted before exit. + @type msg: str + @param trace: flag to output a stack trace before exiting + @type trace: bool + + @return: None + + """ + + retval = int(retval) + trace = bool(trace) + + root_logger = logging.getLogger() + has_handlers = False + if root_logger.handlers: + has_handlers = True + + if msg: + if has_handlers: + if retval: + LOG.error(msg) + else: + LOG.info(msg) + if not has_handlers: + if hasattr(sys.stderr, 'buffer'): + sys.stderr.buffer.write(str(msg) + "\n") + else: + sys.stderr.write(str(msg) + "\n") + + if trace: + if has_handlers: + if retval: + LOG.error(traceback.format_exc()) + else: + LOG.info(traceback.format_exc()) + else: + traceback.print_exc() + + sys.exit(retval) + + # ------------------------------------------------------------------------- + def as_dict(self, short=True): + """ + Transforms the elements of the object into a dict + + @param short: don't include local properties in resulting dict. + @type short: bool + + @return: structure as dict + @rtype: dict + """ + + res = super(CrTplApplication, self).as_dict(short=short) + res['exit_value'] = self.exit_value + res['usage'] = self.usage + res['quiet'] = self.quiet + res['description'] = self.description + res['argparse_epilog'] = self.argparse_epilog + res['argparse_prefix_chars'] = self.argparse_prefix_chars + res['terminal_has_colors'] = self.terminal_has_colors + res['env_prefix'] = self.env_prefix + + return res + + # ------------------------------------------------------------------------- + def init_logging(self): + """ + Initialize the logger object. + It creates a colored loghandler with all output to STDERR. + Maybe overridden in descendant classes. + + @return: None + """ + + log_level = logging.INFO + if self.verbose: + log_level = logging.DEBUG + elif self.quiet: + log_level = logging.WARNING + + root_logger = logging.getLogger() + root_logger.setLevel(log_level) + + # create formatter + format_str = '' + if self.verbose: + format_str = '[%(asctime)s]: ' + format_str += self.appname + ': ' + if self.verbose: + if self.verbose > 1: + format_str += '%(name)s(%(lineno)d) %(funcName)s() ' + else: + format_str += '%(name)s ' + format_str += '%(levelname)s - %(message)s' + formatter = None + if self.terminal_has_colors: + formatter = ColoredFormatter(format_str) + else: + formatter = logging.Formatter(format_str) + + # create log handler for console output + lh_console = logging.StreamHandler(sys.stderr) + lh_console.setLevel(log_level) + lh_console.setFormatter(formatter) + + root_logger.addHandler(lh_console) + + return + + # ------------------------------------------------------------------------- + def terminal_can_color(self): + """ + Method to detect, whether the current terminal (stdout and stderr) + is able to perform ANSI color sequences. + + @return: both stdout and stderr can perform ANSI color sequences + @rtype: bool + + """ + + term_debug = False + if self.verbose > 3: + term_debug = True + return terminal_can_colors(debug=term_debug) + + # ------------------------------------------------------------------------- + def post_init(self): + """ + Method to execute before calling run(). Here could be done some + finishing actions after reading in commandline parameters, + configuration a.s.o. + + This method could be overwritten by descendant classes, these + methhods should allways include a call to post_init() of the + parent class. + + """ + + self.perform_arg_parser() + self.init_logging() + + self.initialized = True + + # ------------------------------------------------------------------------- + def pre_run(self): + """ + Dummy function to run before the main routine. + Could be overwritten by descendant classes. + + """ + + if self.simulate: + LOG.warn("Simulation mode - nothing is really done.") + + # ------------------------------------------------------------------------- + def _run(self): + """ + Dummy function as main routine. + + MUST be overwritten by descendant classes. + + """ + + raise FunctionNotImplementedError('_run', self.__class__.__name__) + + # ------------------------------------------------------------------------- + def __call__(self): + """ + Helper method to make the resulting object callable, e.g.:: + + app = PBApplication(...) + app() + + @return: None + + """ + + self.run() + + # ------------------------------------------------------------------------- + def run(self): + """ + The visible start point of this object. + + @return: None + + """ + + if not self.initialized: + self.handle_error( + "The application is not completely initialized.", '', True) + self.exit(9) + + try: + self.pre_run() + except Exception as e: + self.handle_error(str(e), e.__class__.__name__, True) + self.exit(98) + + if not self.initialized: + raise PpAppError( + "Object {!r} seems not to be completely initialized.".format( + self.__class__.__name__)) + + try: + self._run() + except Exception as e: + self.handle_error(str(e), e.__class__.__name__, True) + self.exit_value = 99 + + if self.verbose > 1: + LOG.info("Ending.") + + try: + self.post_run() + except Exception as e: + self.handle_error(str(e), e.__class__.__name__, True) + self.exit_value = 97 + + self.exit(self.exit_value) + + # ------------------------------------------------------------------------- + def post_run(self): + """ + Dummy function to run after the main routine. + Could be overwritten by descendant classes. + + """ + + if self.verbose > 1: + LOG.info("executing post_run() ...") + + # ------------------------------------------------------------------------- + def _init_arg_parser(self): + """ + Local called method to initiate the argument parser. + + @raise PBApplicationError: on some errors + + """ + + self.arg_parser = argparse.ArgumentParser( + prog=self.appname, + description=self.description, + usage=self.usage, + epilog=self.argparse_epilog, + prefix_chars=self.argparse_prefix_chars, + add_help=False, + ) + + self.init_arg_parser() + + general_group = self.arg_parser.add_argument_group('General options') + general_group.add_argument( + '--color', + action="store", + dest='color', + const='yes', + default='auto', + nargs='?', + choices=['yes', 'no', 'auto'], + help="Use colored output for messages.", + ) + + verbose_group = general_group.add_mutually_exclusive_group() + + verbose_group.add_argument( + "-v", "--verbose", + action="count", + dest='verbose', + help='Increase the verbosity level', + ) + + verbose_group.add_argument( + "-q", "--quiet", + action="store_true", + dest='quiet', + help='Silent execution, only warnings and errors are emitted.', + ) + + general_group.add_argument( + "-h", "--help", + action='help', + dest='help', + help='Show this help message and exit' + ) + general_group.add_argument( + "--usage", + action='store_true', + dest='usage', + help="Display brief usage message and exit" + ) + general_group.add_argument( + "-V", '--version', + action='version', + version='Version of %(prog)s: {}'.format(self.version), + help="Show program's version number and exit" + ) + + # ------------------------------------------------------------------------- + def init_arg_parser(self): + """ + Public available method to initiate the argument parser. + + Note:: + avoid adding the general options '--verbose', '--help', '--usage' + and '--version'. These options are allways added after executing + this method. + + Descendant classes may override this method. + + """ + + pass + + # ------------------------------------------------------------------------- + def _perform_arg_parser(self): + """ + Underlaying method for parsing arguments. + """ + + self.args = self.arg_parser.parse_args() + + if self.args.usage: + self.arg_parser.print_usage(sys.stdout) + self.exit(0) + + if self.args.verbose is not None and self.args.verbose > self.verbose: + self.verbose = self.args.verbose + + if self.args.quiet: + self.quiet = self.args.quiet + + if self.args.color == 'yes': + self._terminal_has_colors = True + elif self.args.color == 'no': + self._terminal_has_colors = False + else: + self._terminal_has_colors = self.terminal_can_color() + + # ------------------------------------------------------------------------- + def perform_arg_parser(self): + """ + Public available method to execute some actions after parsing + the command line parameters. + + Descendant classes may override this method. + """ + + pass + + # ------------------------------------------------------------------------- + def _init_env(self): + """ + Initialization of self.env by application specific environment + variables. + + It calls self.init_env(), after it has done his job. + + """ + + for (key, value) in list(os.environ.items()): + + if not key.startswith(self.env_prefix): + continue + + newkey = key.replace(self.env_prefix, '', 1) + self.env[newkey] = value + + self.init_env() + + # ------------------------------------------------------------------------- + def init_env(self): + """ + Public available method to initiate self.env additional to the implicit + initialization done by this module. + Maybe it can be used to import environment variables, their + names not starting with self.env_prefix. + + Currently a dummy method, which ca be overriden by descendant classes. + + """ + + pass + + # ------------------------------------------------------------------------- + def _perform_env(self): + """ + Method to do some useful things with the found environment. + + It calls self.perform_env(), after it has done his job. + + """ + + # try to detect verbosity level from environment + if 'VERBOSE' in self.env and self.env['VERBOSE']: + v = 0 + try: + v = int(self.env['VERBOSE']) + except ValueError: + v = 1 + if v > self.verbose: + self.verbose = v + + self.perform_env() + + # ------------------------------------------------------------------------- + def perform_env(self): + """ + Public available method to perform found environment variables after + initialization of self.env. + + Currently a dummy method, which ca be overriden by descendant classes. + + """ + + pass + + # ------------------------------------------------------------------------- + def colored(self, msg, color): + """ + Wrapper function to colorize the message. Depending, whether the current + terminal can display ANSI colors, the message is colorized or not. + + @param msg: The message to colorize + @type msg: str + @param color: The color to use, must be one of the keys of COLOR_CODE + @type color: str + + @return: the colorized message + @rtype: str + + """ + + if not self.terminal_has_colors: + return msg + return colorstr(msg, color) + + # ------------------------------------------------------------------------- + def get_command(self, cmd, quiet=False): + """ + Searches the OS search path for the given command and gives back the + normalized position of this command. + If the command is given as an absolute path, it check the existence + of this command. + + @param cmd: the command to search + @type cmd: str + @param quiet: No warning message, if the command could not be found, + only a debug message + @type quiet: bool + + @return: normalized complete path of this command, or None, + if not found + @rtype: str or None + + """ + + if self.verbose > 2: + LOG.debug("Searching for command {!r} ...".format(cmd)) + + # Checking an absolute path + if os.path.isabs(cmd): + if not os.path.exists(cmd): + LOG.warning("Command {!r} doesn't exists.".format(cmd)) + return None + if not os.access(cmd, os.X_OK): + msg = "Command {!r} is not executable.".format(cmd) + LOG.warning(msg) + return None + return os.path.normpath(cmd) + + # Checking a relative path + for d in caller_search_path(): + if self.verbose > 3: + LOG.debug("Searching command in {!r} ...".format(d)) + p = os.path.join(d, cmd) + if os.path.exists(p): + if self.verbose > 2: + LOG.debug("Found {!r} ...".format(p)) + if os.access(p, os.X_OK): + return os.path.normpath(p) + else: + LOG.debug("Command {!r} is not executable.".format(p)) + + # command not found, sorry + if quiet: + if self.verbose > 2: + LOG.debug("Command {!r} not found.".format(cmd)) + else: + LOG.warning("Command {!r} not found.".format(cmd)) + + return None + +# ============================================================================= + +if __name__ == "__main__": + + pass + +# ============================================================================= + +# vim: tabstop=4 expandtab shiftwidth=4 softtabstop=4 list diff --git a/lib/cr_vmware_tpl/colored.py b/lib/cr_vmware_tpl/colored.py new file mode 100644 index 0000000..97bbfe6 --- /dev/null +++ b/lib/cr_vmware_tpl/colored.py @@ -0,0 +1,215 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +""" +@summary: additional logging formatter for colored output via console +""" + +# Standard modules +import logging +import copy + +# Third party modules + +# Own modules + +__version__ = '0.1.1' + +# ============================================================================= +# Color coding module variables and helper functions + +COLOR_CODE = { + 'ENDC': 0, # RESET COLOR + 'BOLD': 1, + 'UNDERLINE': 4, + 'BLINK': 5, + 'INVERT': 7, + 'CONCEALD': 8, + 'STRIKE': 9, + 'GREY30': 90, + 'GREY40': 2, + 'GREY65': 37, + 'GREY70': 97, + 'GREY20_BG': 40, + 'GREY33_BG': 100, + 'GREY80_BG': 47, + 'GREY93_BG': 107, + 'DARK_RED': 31, + 'RED': 91, + 'RED_BG': 41, + 'LIGHT_RED_BG': 101, + 'DARK_YELLOW': 33, + 'YELLOW': 93, + 'YELLOW_BG': 43, + 'LIGHT_YELLOW_BG': 103, + 'DARK_BLUE': 34, + 'BLUE': 94, + 'BLUE_BG': 44, + 'LIGHT_BLUE_BG': 104, + 'DARK_MAGENTA': 35, + 'PURPLE': 95, + 'MAGENTA_BG': 45, + 'LIGHT_PURPLE_BG': 105, + 'DARK_CYAN': 36, + 'AUQA': 96, + 'AQUA': 96, + 'CYAN_BG': 46, + 'LIGHT_AUQA_BG': 106, + 'LIGHT_AQUA_BG': 106, + 'DARK_GREEN': 32, + 'GREEN': 92, + 'GREEN_BG': 42, + 'LIGHT_GREEN_BG': 102, + 'BLACK': 30, +} + + +# ----------------------------------------------------------------------------- +def termcode(num): + """ + Output of an ANSII terminal code. + """ + + return('\033[%sm' % (num)) + + +# ----------------------------------------------------------------------------- +def colorstr(message, color): + """ + Wrapper function to colorize the message. + + @param message: The message to colorize + @type message: str + @param color: The color to use, must be one of the keys of COLOR_CODE + @type color: str + + @return: the colorized message + @rtype: str + + """ + + tcode = '' + if isinstance(color, (list, tuple)): + for clr in color: + tcode += termcode(COLOR_CODE[clr]) + else: + tcode = termcode(COLOR_CODE[color]) + + return tcode + message + termcode(COLOR_CODE['ENDC']) + + +# ---------------------------------------- +LOG = logging.getLogger(__name__) + + +# ============================================================================= +class ColoredFormatter(logging.Formatter): + """ + A variant of code found at: + http://stackoverflow.com/questions/384076/how-can-i-make-the-python-logging-output-to-be-colored + """ + + LEVEL_COLOR = { + 'DEBUG': None, + 'INFO': 'GREEN', + 'WARNING': 'YELLOW', + 'ERROR': ('BOLD', 'RED'), + 'CRITICAL': 'RED_BG', + } + + # ------------------------------------------------------------------------- + def __init__(self, fmt=None, datefmt=None): + """ + Initialize the formatter with specified format strings. + + Initialize the formatter either with the specified format string, or a + default. Allow for specialized date formatting with the optional + datefmt argument (if omitted, you get the ISO8601 format). + """ + + logging.Formatter.__init__(self, fmt, datefmt) + + # ----------------------------------------------------------- + @property + def color_debug(self): + """The color used to output debug messages.""" + return self.LEVEL_COLOR['DEBUG'] + + @color_debug.setter + def color_debug(self, value): + self.LEVEL_COLOR['DEBUG'] = value + + # ----------------------------------------------------------- + @property + def color_info(self): + """The color used to output info messages.""" + return self.LEVEL_COLOR['INFO'] + + @color_info.setter + def color_info(self, value): + self.LEVEL_COLOR['INFO'] = value + + # ----------------------------------------------------------- + @property + def color_warning(self): + """The color used to output warning messages.""" + return self.LEVEL_COLOR['WARNING'] + + @color_warning.setter + def color_warning(self, value): + self.LEVEL_COLOR['WARNING'] = value + + # ----------------------------------------------------------- + @property + def color_error(self): + """The color used to output error messages.""" + return self.LEVEL_COLOR['ERROR'] + + @color_error.setter + def color_error(self, value): + self.LEVEL_COLOR['ERROR'] = value + + # ----------------------------------------------------------- + @property + def color_critical(self): + """The color used to output critical messages.""" + return self.LEVEL_COLOR['CRITICAL'] + + @color_critical.setter + def color_critical(self, value): + self.LEVEL_COLOR['CRITICAL'] = value + + # ------------------------------------------------------------------------- + def format(self, record): + """ + Format the specified record as text. + """ + + record = copy.copy(record) + levelname = record.levelname + + if levelname in self.LEVEL_COLOR: + + record.name = colorstr(record.name, 'BOLD') + record.filename = colorstr(record.filename, 'BOLD') + record.module = colorstr(record.module, 'BOLD') + record.funcName = colorstr(record.funcName, 'BOLD') + record.pathname = colorstr(record.pathname, 'BOLD') + record.processName = colorstr(record.processName, 'BOLD') + record.threadName = colorstr(record.threadName, 'BOLD') + + if self.LEVEL_COLOR[levelname] is not None: + record.levelname = colorstr( + levelname, self.LEVEL_COLOR[levelname]) + record.msg = colorstr(record.msg, self.LEVEL_COLOR[levelname]) + + return logging.Formatter.format(self, record) + + +# ============================================================================= + +if __name__ == "__main__": + pass + +# ============================================================================= + +# vim: tabstop=4 expandtab shiftwidth=4 softtabstop=4 diff --git a/lib/cr_vmware_tpl/errors.py b/lib/cr_vmware_tpl/errors.py new file mode 100644 index 0000000..4dad12c --- /dev/null +++ b/lib/cr_vmware_tpl/errors.py @@ -0,0 +1,153 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +""" +@author: Frank Brehm +@summary: module for some common used error classes +""" + +# Standard modules +import errno + + +__version__ = '0.1.1' + +# ============================================================================= +class PpError(Exception): + """ + Base error class for all other self defined exceptions. + """ + + pass + + +# ============================================================================= +class PpAppError(PpError): + + pass + + +# ============================================================================= +class FunctionNotImplementedError(PpError, NotImplementedError): + """ + Error class for not implemented functions. + """ + + # ------------------------------------------------------------------------- + def __init__(self, function_name, class_name): + """ + Constructor. + + @param function_name: the name of the not implemented function + @type function_name: str + @param class_name: the name of the class of the function + @type class_name: str + + """ + + self.function_name = function_name + if not function_name: + self.function_name = '__unkown_function__' + + self.class_name = class_name + if not class_name: + self.class_name = '__unkown_class__' + + # ------------------------------------------------------------------------- + def __str__(self): + """ + Typecasting into a string for error output. + """ + + msg = "Function {func}() has to be overridden in class {cls!r}." + return msg.format(func=self.function_name, cls=self.class_name) + + +# ============================================================================= +class IoTimeoutError(PpError, IOError): + """ + Special error class indicating a timout error on a read/write operation + """ + + # ------------------------------------------------------------------------- + def __init__(self, strerror, timeout, filename=None): + """ + Constructor. + + @param strerror: the error message about the operation + @type strerror: str + @param timeout: the timout in seconds leading to the error + @type timeout: float + @param filename: the filename leading to the error + @type filename: str + + """ + + t_o = None + try: + t_o = float(timeout) + except ValueError: + pass + self.timeout = t_o + + if t_o is not None: + strerror += " (timeout after {:0.1f} secs)".format(t_o) + + if filename is None: + super(IoTimeoutError, self).__init__(errno.ETIMEDOUT, strerror) + else: + super(IoTimeoutError, self).__init__( + errno.ETIMEDOUT, strerror, filename) + + +# ============================================================================= +class ReadTimeoutError(IoTimeoutError): + """ + Special error class indicating a timout error on reading of a file. + """ + + # ------------------------------------------------------------------------- + def __init__(self, timeout, filename): + """ + Constructor. + + @param timeout: the timout in seconds leading to the error + @type timeout: float + @param filename: the filename leading to the error + @type filename: str + + """ + + strerror = "Timeout error on reading" + super(ReadTimeoutError, self).__init__(strerror, timeout, filename) + + +# ============================================================================= +class WriteTimeoutError(IoTimeoutError): + """ + Special error class indicating a timout error on a writing into a file. + """ + + # ------------------------------------------------------------------------- + def __init__(self, timeout, filename): + """ + Constructor. + + @param timeout: the timout in seconds leading to the error + @type timeout: float + @param filename: the filename leading to the error + @type filename: str + + """ + + strerror = "Timeout error on writing" + super(WriteTimeoutError, self).__init__(strerror, timeout, filename) + + +# ============================================================================= + +if __name__ == "__main__": + pass + +# ============================================================================= + +# vim: tabstop=4 expandtab shiftwidth=4 softtabstop=4 diff --git a/lib/cr_vmware_tpl/obj.py b/lib/cr_vmware_tpl/obj.py new file mode 100644 index 0000000..8208c98 --- /dev/null +++ b/lib/cr_vmware_tpl/obj.py @@ -0,0 +1,328 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +""" +@author: Frank Brehm +@contact: frank.brehm@pixelpark.com +@copyright: © 2018 by Frank Brehm, Publicies Pixelpark GmbH, Berlin +""" +from __future__ import absolute_import + +# Standard modules +import sys +import os +import logging +import datetime +import traceback + +# Third party modules + +# Own modules +from .common import pp, to_bytes + +from .errors import PpError + +__version__ = '0.2.4' + +LOG = logging.getLogger(__name__) + + +# ============================================================================= +class PpBaseObjectError(PpError): + """ + Base error class useable by all descendand objects. + """ + + pass + + +# ============================================================================= +class PpBaseObject(object): + """ + Base class for all objects. + """ + + # ------------------------------------------------------------------------- + def __init__( + self, appname=None, verbose=0, version=__version__, base_dir=None, + initialized=False): + """ + Initialisation of the base object. + + Raises an exception on a uncoverable error. + """ + + self._appname = None + """ + @ivar: name of the current running application + @type: str + """ + if appname: + v = str(appname).strip() + if v: + self._appname = v + if not self._appname: + self._appname = os.path.basename(sys.argv[0]) + + self._version = version + """ + @ivar: version string of the current object or application + @type: str + """ + + self._verbose = int(verbose) + """ + @ivar: verbosity level (0 - 9) + @type: int + """ + if self._verbose < 0: + msg = "Wrong verbose level {!r}, must be >= 0".format(verbose) + raise ValueError(msg) + + self._initialized = False + """ + @ivar: initialisation of this object is complete + after __init__() of this object + @type: bool + """ + + self._base_dir = base_dir + """ + @ivar: base directory used for different purposes, must be an existent + directory. Defaults to directory of current script daemon.py. + @type: str + """ + if base_dir: + if not os.path.exists(base_dir): + msg = "Base directory {!r} does not exists.".format(base_dir) + self.handle_error(msg) + self._base_dir = None + elif not os.path.isdir(base_dir): + msg = "Base directory {!r} is not a directory.".format(base_dir) + self.handle_error(msg) + self._base_dir = None + if not self._base_dir: + self._base_dir = os.path.dirname(os.path.abspath(sys.argv[0])) + + self._initialized = bool(initialized) + + # ----------------------------------------------------------- + @property + def appname(self): + """The name of the current running application.""" + if hasattr(self, '_appname'): + return self._appname + return os.path.basename(sys.argv[0]) + + @appname.setter + def appname(self, value): + if value: + v = str(value).strip() + if v: + self._appname = v + + # ----------------------------------------------------------- + @property + def version(self): + """The version string of the current object or application.""" + return getattr(self, '_version', __version__) + + # ----------------------------------------------------------- + @property + def verbose(self): + """The verbosity level.""" + return getattr(self, '_verbose', 0) + + @verbose.setter + def verbose(self, value): + v = int(value) + if v >= 0: + self._verbose = v + else: + LOG.warn("Wrong verbose level {!r}, must be >= 0".format(value)) + + # ----------------------------------------------------------- + @property + def initialized(self): + """The initialisation of this object is complete.""" + return getattr(self, '_initialized', False) + + @initialized.setter + def initialized(self, value): + self._initialized = bool(value) + + # ----------------------------------------------------------- + @property + def base_dir(self): + """The base directory used for different purposes.""" + return self._base_dir + + @base_dir.setter + def base_dir(self, value): + if value.startswith('~'): + value = os.path.expanduser(value) + if not os.path.exists(value): + msg = "Base directory {!r} does not exists.".format(value) + LOG.error(msg) + elif not os.path.isdir(value): + msg = "Base directory {!r} is not a directory.".format(value) + LOG.error(msg) + else: + self._base_dir = value + + # ------------------------------------------------------------------------- + def __str__(self): + """ + Typecasting function for translating object structure + into a string + + @return: structure as string + @rtype: str + """ + + return pp(self.as_dict(short=True)) + + # ------------------------------------------------------------------------- + def __repr__(self): + """Typecasting into a string for reproduction.""" + + out = "<%s(" % (self.__class__.__name__) + + fields = [] + fields.append("appname={!r}".format(self.appname)) + fields.append("verbose={!r}".format(self.verbose)) + fields.append("version={!r}".format(self.version)) + fields.append("base_dir={!r}".format(self.base_dir)) + fields.append("initialized={!r}".format(self.initialized)) + + out += ", ".join(fields) + ")>" + return out + + # ------------------------------------------------------------------------- + def as_dict(self, short=True): + """ + Transforms the elements of the object into a dict + + @param short: don't include local properties in resulting dict. + @type short: bool + + @return: structure as dict + @rtype: dict + """ + + res = self.__dict__ + res = {} + for key in self.__dict__: + if short and key.startswith('_') and not key.startswith('__'): + continue + val = self.__dict__[key] + if isinstance(val, PpBaseObject): + res[key] = val.as_dict(short=short) + else: + res[key] = val + res['__class_name__'] = self.__class__.__name__ + res['appname'] = self.appname + res['version'] = self.version + res['verbose'] = self.verbose + res['initialized'] = self.initialized + res['base_dir'] = self.base_dir + + return res + + # ------------------------------------------------------------------------- + def handle_error( + self, error_message=None, exception_name=None, do_traceback=False): + """ + Handle an error gracefully. + + Print a traceback and continue. + + @param error_message: the error message to display + @type error_message: str + @param exception_name: name of the exception class + @type exception_name: str + @param do_traceback: allways show a traceback + @type do_traceback: bool + + """ + + msg = 'Exception happened: ' + if exception_name is not None: + exception_name = exception_name.strip() + if exception_name: + msg = exception_name + ': ' + else: + msg = '' + if error_message: + msg += str(error_message) + else: + msg += 'undefined error.' + + root_log = logging.getLogger() + has_handlers = False + if root_log.handlers: + has_handlers = True + + if has_handlers: + LOG.error(msg) + if do_traceback: + LOG.error(traceback.format_exc()) + else: + curdate = datetime.datetime.now() + curdate_str = "[" + curdate.isoformat(' ') + "]: " + msg = curdate_str + msg + "\n" + if hasattr(sys.stderr, 'buffer'): + sys.stderr.buffer.write(to_bytes(msg)) + else: + sys.stderr.write(msg) + if do_traceback: + traceback.print_exc() + + return + + # ------------------------------------------------------------------------- + def handle_info(self, message, info_name=None): + """ + Shows an information. This happens both to STDERR and to all + initialized log handlers. + + @param message: the info message to display + @type message: str + @param info_name: Title of information + @type info_name: str + + """ + + msg = '' + if info_name is not None: + info_name = info_name.strip() + if info_name: + msg = info_name + ': ' + msg += str(message).strip() + + root_log = logging.getLogger() + has_handlers = False + if root_log.handlers: + has_handlers = True + + if has_handlers: + LOG.info(msg) + else: + curdate = datetime.datetime.now() + curdate_str = "[" + curdate.isoformat(' ') + "]: " + msg = curdate_str + msg + "\n" + if hasattr(sys.stderr, 'buffer'): + sys.stderr.buffer.write(to_bytes(msg)) + else: + sys.stderr.write(msg) + + return + +# ============================================================================= + +if __name__ == "__main__": + + pass + +# ============================================================================= + +# vim: tabstop=4 expandtab shiftwidth=4 softtabstop=4 -- 2.39.5