Add CustomFormatter class colouring logging output

Change pre-logger criticals to format strings and add colouring
Add --no-color option to disable above
Reduce number of statements in main.parse_options

source/main.py

@@ -6,21 +6,22 @@
 import datetime
 import errno
 import pkg_resources
-from ansible_deployer.modules.globalvars import SUBCOMMANDS
 from ansible_deployer.modules.configs.config import Config
 from ansible_deployer.modules.locking.locking import Locking
 from ansible_deployer.modules.outputs.logging import Loggers
 from ansible_deployer.modules.validators.validate import Validators
 from ansible_deployer.modules.runners.run import Runners
 from ansible_deployer.modules import misc
+from ansible_deployer.modules import globalvars
 
 def parse_options(argv):
     """Generic function to parse options for all commands, we validate if the option was allowed for
     specific subcommand outside"""
     parser = argparse.ArgumentParser(add_help=True)
 
     parser.add_argument("subcommand", nargs='*', default=None, metavar="SUBCOMMAND",
-                        help='Specify subcommand to execute. Available commands: '+str(SUBCOMMANDS))
+                        help='Specify subcommand to execute. Available commands: '+ \
+                        str(globalvars.SUBCOMMANDS))
     parser.add_argument("--infrastructure", "-i", nargs=1, default=[None], metavar="INFRASTRUCTURE",
                         help='Specify infrastructure for deploy.')
     parser.add_argument("--stage", "-s", nargs=1, default=[None], metavar="STAGE",
@@ -49,6 +50,8 @@ def parse_options(argv):
                         help='Setup repo outside of workdir in requested path. This option applies'
                         ' only to infrastructures with allow_user_checkout enabled in infra'
                         ' config!')
+    parser.add_argument("--no-color", default=False, action="store_true", help='Disable coloring'
+                        'of console messages.')
 
     arguments = parser.parse_args(argv)
 
@@ -57,16 +60,22 @@ def parse_options(argv):
         print(f"ansible-deployer version: {version}")
         sys.exit(0)
 
+    if arguments.no_color:
+        PRINT_FAIL = PRINT_END = ""
+    else:
+        PRINT_FAIL = globalvars.PRINT_FAIL
+        PRINT_END = globalvars.PRINT_END
+
     if not arguments.subcommand:
-        sub_string = ", ".join(SUBCOMMANDS).strip(", ")
-        print(f"[CRITICAL]: First positional argument (subcommand) is required! Available commands"
-              f" are: {sub_string}.")
+        sub_string = ", ".join(globalvars.SUBCOMMANDS).strip(", ")
+        print(f"{PRINT_FAIL}[CRITICAL]: First positional argument (subcommand) is required!"
+              f" Available commands are: {sub_string}.{PRINT_END}")
         sys.exit(57)
 
     options = {}
     options["subcommand"] = arguments.subcommand[0].lower()
-    Validators.verify_subcommand(options["subcommand"])
-    Validators.verify_switches(arguments.subcommand)
+    Validators.verify_subcommand(options["subcommand"], arguments.no_color)
+    Validators.verify_switches(arguments.subcommand, arguments.no_color)
 
     options["switches"] = arguments.subcommand[1:]
     options["infra"] = arguments.infrastructure[0]
@@ -79,14 +88,10 @@ def parse_options(argv):
     options["syslog"] = arguments.syslog
     options["limit"] = arguments.limit[0]
     options["raw_output"] = arguments.raw_runner_output
-    if arguments.self_setup[0]:
-        options["self_setup"] = os.path.abspath(arguments.self_setup[0])
-    else:
-        options["self_setup"] = None
-    if arguments.conf_dir[0]:
-        options["conf_dir"] = os.path.abspath(arguments.conf_dir[0])
-    else:
-        options["conf_dir"] = None
+    options["self_setup"] = os.path.abspath(arguments.self_setup[0]) if arguments.self_setup[0] \
+                            else None
+    options["conf_dir"] = os.path.abspath(arguments.conf_dir[0]) if arguments.conf_dir[0] else None
+    options["no_color"] = arguments.no_color
 
     return options
 
@@ -95,7 +100,8 @@ def main():
     start_ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
 
     if len(sys.argv) < 2:
-        print("[CRITICAL]: Too few arguments", file=sys.stderr)
+        print(f"{globalvars.PRINT_FAIL}[CRITICAL]: Too few arguments{globalvars.PRINT_END}",
+              file=sys.stderr)
         sys.exit(2)
     options = parse_options(sys.argv[1:])
 

source/modules/globalvars.py

@@ -3,3 +3,5 @@
 APP_CONF = "/etc/ansible-deployer"
 CFG_PERMISSIONS = "0o644"
 SUBCOMMANDS = ("run", "lock", "unlock", "verify", "show")
+PRINT_FAIL = '\033[91m'
+PRINT_END = '\033[0m'

source/modules/outputs/logging.py

@@ -17,7 +17,7 @@ def set_logging(options: dict):
         logger = logging.getLogger("ansible-deployer_log")
         logger.setLevel(logging.DEBUG)
         formatter = logging.Formatter("%(asctime)s [%(levelname)s]: %(message)s")
-        console_formatter = logging.Formatter("\n%(asctime)s [%(levelname)s]: %(message)s\n")
+        console_formatter = "\n%(asctime)s [%(levelname)s]: %(message)s\n"
 
         if options["syslog"]:
             rsys_handler = log_han.SysLogHandler(address="/dev/log")
@@ -31,7 +31,8 @@ def set_logging(options: dict):
         logger.addHandler(memory_handler)
 
         console_handler = logging.StreamHandler()
-        console_handler.setFormatter(console_formatter)
+        console_handler.setFormatter(logging.Formatter(console_formatter) if options["no_color"] \
+                                     else CustomFormatter(console_formatter))
         console_handler.setLevel(logging.DEBUG if options["debug"] else logging.INFO)
         logger.addHandler(console_handler)
 
@@ -46,3 +47,29 @@ def set_logging_to_file(self, log_dir: str, timestamp: str, conf: dict):
         file_handler.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s]: %(message)s"))
         file_handler.setLevel(logging.DEBUG)
         self.logger.addHandler(file_handler)
+
+
+class CustomFormatter(logging.Formatter):
+    """Class adding colours to console logger"""
+
+    def __init__(self, formatter):
+        super().__init__()
+        grey = "\x1b[0;38m"
+        light_green = "\x1b[1;32m"
+        yellow = "\x1b[0;33m"
+        red = "\x1b[0;31m"
+        light_red = "\x1b[1;31m"
+        reset = "\x1b[0m"
+
+        self.FORMATS = {
+            logging.DEBUG: light_green + formatter + reset,
+            logging.INFO: grey + formatter + reset,
+            logging.WARNING: yellow + formatter + reset,
+            logging.ERROR: red + formatter + reset,
+            logging.CRITICAL: light_red + formatter + reset
+        }
+
+    def format(self, record):
+        log_fmt = self.FORMATS.get(record.levelno)
+        formatter = logging.Formatter(log_fmt)
+        return formatter.format(record)

source/modules/validators/validate.py

@@ -3,7 +3,7 @@
 import sys
 import os
 import re
-from ansible_deployer.modules.globalvars import SUBCOMMANDS
+from ansible_deployer.modules import globalvars
 
 
 class Validators:
@@ -13,26 +13,39 @@ def __init__(self, logger):
         self.logger = logger
 
     @staticmethod
-    def verify_subcommand(command: str):
+    def verify_subcommand(command: str, color_flag: bool):
         """Function to check the first arguments for a valid subcommand"""
-        if command not in SUBCOMMANDS:
-            print("[CRITICAL]: Unknown subcommand :%s", (command), file=sys.stderr)
+        if color_flag:
+            PRINT_FAIL = PRINT_END = ""
+        else:
+            PRINT_FAIL = globalvars.PRINT_FAIL
+            PRINT_END = globalvars.PRINT_END
+
+        if command not in globalvars.SUBCOMMANDS:
+            print(f"{PRINT_FAIL}[CRITICAL]: Unknown subcommand :%s {PRINT_END}", (command),
+                  file=sys.stderr)
             sys.exit("55")
 
     @staticmethod
-    def verify_switches(switches: list):
+    def verify_switches(switches: list, color_flag: bool):
         """
         Check if 2nd and following positional arguments are valid
         """
+        if color_flag:
+            PRINT_FAIL = PRINT_END = ""
+        else:
+            PRINT_FAIL = globalvars.PRINT_FAIL
+            PRINT_END = globalvars.PRINT_END
+
         if switches[0] != "show" and len(switches[1:]) > 0:
-            print("[CRITICAL]: Too many positional arguments! Only subcommand \"show\" can accept"
-                  " following arguments: all, task, infra.")
+            print(f"{PRINT_FAIL}[CRITICAL]: Too many positional arguments! Only subcommand \"show\""
+                  f" can accept following arguments: all, task, infra.{PRINT_END}")
             sys.exit("56")
 
         for switch in switches[1:]:
-            if switch not in ("all", "task", "infra"):
-                print(f"[CRITICAL]: Invalid argument {switch}! Subcommand \"show\" can accept only"
-                      " following arguments: all, task, infra.")
+            if switch not in ("task", "infra"):
+                print(f"{PRINT_FAIL}[CRITICAL]: Invalid argument {switch}! Subcommand \"show\" can"
+                      f" accept only following arguments: all, task, infra.{PRINT_END}")
                 sys.exit("57")
 
     def validate_options(self, options: dict):