diff libervia/backend/core/log.py @ 4270:0d7bb4df2343

Reformatted code base using black.
author Goffi <goffi@goffi.org>
date Wed, 19 Jun 2024 18:44:57 +0200
parents 4b842c1fb686
children
line wrap: on
line diff
--- a/libervia/backend/core/log.py	Tue Jun 18 12:06:45 2024 +0200
+++ b/libervia/backend/core/log.py	Wed Jun 19 18:44:57 2024 +0200
@@ -36,8 +36,8 @@
 backend = None
 _loggers: Dict[str, "Logger"] = {}
 handlers = {}
-COLOR_START = '%(color_start)s'
-COLOR_END = '%(color_end)s'
+COLOR_START = "%(color_start)s"
+COLOR_END = "%(color_end)s"
 
 
 class Filtered(Exception):
@@ -46,8 +46,9 @@
 
 class Logger:
     """High level logging class"""
-    fmt = None # format option as given by user (e.g. SAT_LOG_LOGGER)
-    filter_name = None # filter to call
+
+    fmt = None  # format option as given by user (e.g. SAT_LOG_LOGGER)
+    filter_name = None  # filter to call
     post_treat = None
 
     def __init__(self, name):
@@ -72,7 +73,7 @@
         message: object,
         level: Optional[str] = None,
         exc_info: _ExcInfoType = False,
-        **kwargs
+        **kwargs,
     ) -> None:
         """Actually log the message
 
@@ -83,11 +84,7 @@
         print(message)
 
     def log(
-        self,
-        level: str,
-        message: object,
-        exc_info: _ExcInfoType = False,
-        **kwargs
+        self, level: str, message: object, exc_info: _ExcInfoType = False, **kwargs
     ) -> None:
         """Print message
 
@@ -116,14 +113,18 @@
         """
         if self.fmt is None and self.filter_name is None:
             return message
-        record = {'name': self._name,
-                  'message': message,
-                  'levelname': level,
-                 }
+        record = {
+            "name": self._name,
+            "message": message,
+            "levelname": level,
+        }
         try:
             if not self.filter_name.dict_filter(record):
                 raise Filtered
-        except (AttributeError, TypeError): # XXX: TypeError is here because of a pyjamas bug which need to be fixed (TypeError is raised instead of AttributeError)
+        except (
+            AttributeError,
+            TypeError,
+        ):  # XXX: TypeError is here because of a pyjamas bug which need to be fixed (TypeError is raised instead of AttributeError)
             if self.filter_name is not None:
                 raise ValueError("Bad filter: filters must have a .filter method")
         try:
@@ -131,9 +132,9 @@
         except TypeError:
             return message
         except KeyError as e:
-            if e.args[0] == 'profile':
+            if e.args[0] == "profile":
                 # XXX: %(profile)s use some magic with introspection, for debugging purpose only *DO NOT* use in production
-                record['profile'] = configure_cls[backend].get_profile()
+                record["profile"] = configure_cls[backend].get_profile()
                 return self.fmt % record
             else:
                 raise e
@@ -167,6 +168,7 @@
         """
         assert name_re
         import re
+
         self.name_re = re.compile(name_re)
 
     def filter(self, record):
@@ -180,10 +182,12 @@
         @param dict_record: dictionary with at list a key "name" with logger name
         @return: True if message should be logged
         """
+
         class LogRecord(object):
             pass
+
         log_record = LogRecord()
-        log_record.name = dict_record['name']
+        log_record.name = dict_record["name"]
         return self.filter(log_record) == 1
 
 
@@ -192,8 +196,17 @@
     # True if color location is specified in fmt (with COLOR_START)
     _color_location = False
 
-    def __init__(self, level=None, fmt=None, output=None, logger=None, colors=False,
-                 levels_taints_dict=None, force_colors=False, backend_data=None):
+    def __init__(
+        self,
+        level=None,
+        fmt=None,
+        output=None,
+        logger=None,
+        colors=False,
+        levels_taints_dict=None,
+        force_colors=False,
+        backend_data=None,
+    ):
         """Configure a backend
 
         @param level: one of C.LOG_LEVELS
@@ -231,19 +244,21 @@
         if level is not None:
             # we deactivate methods below level
             level_idx = C.LOG_LEVELS.index(level)
+
             def dev_null(self, msg):
                 pass
+
             for _level in C.LOG_LEVELS[:level_idx]:
                 setattr(Logger, _level.lower(), dev_null)
 
     def configure_format(self, fmt):
         if fmt is not None:
-            if fmt != '%(message)s': # %(message)s is the same as None
+            if fmt != "%(message)s":  # %(message)s is the same as None
                 Logger.fmt = fmt
             if COLOR_START in fmt:
                 ConfigureBase._color_location = True
-                if fmt.find(COLOR_END,fmt.rfind(COLOR_START))<0:
-                   # color_start not followed by an end, we add it
+                if fmt.find(COLOR_END, fmt.rfind(COLOR_START)) < 0:
+                    # color_start not followed by an end, we add it
                     Logger.fmt += COLOR_END
 
     def configure_output(self, output):
@@ -265,12 +280,14 @@
             taints = self.__class__.taints = {}
             for level in C.LOG_LEVELS:
                 # we want use values and use constant value as default
-                taint_list = levels_taints_dict.get(level, C.LOG_OPT_TAINTS_DICT[1][level])
+                taint_list = levels_taints_dict.get(
+                    level, C.LOG_OPT_TAINTS_DICT[1][level]
+                )
                 ansi_list = []
                 for elt in taint_list:
                     elt = elt.upper()
                     try:
-                        ansi = getattr(A, 'FG_{}'.format(elt))
+                        ansi = getattr(A, "FG_{}".format(elt))
                     except AttributeError:
                         try:
                             ansi = getattr(A, elt)
@@ -278,13 +295,13 @@
                             # we use raw string if element is unknown
                             ansi = elt
                     ansi_list.append(ansi)
-                taints[level] = ''.join(ansi_list)
+                taints[level] = "".join(ansi_list)
 
     def post_treatment(self):
         pass
 
     def manage_outputs(self, outputs_raw):
-        """ Parse output option in a backend agnostic way, and fill handlers consequently
+        """Parse output option in a backend agnostic way, and fill handlers consequently
 
         @param outputs_raw: output option as enterred in environment variable or in configuration
         """
@@ -298,15 +315,19 @@
         for output in outputs:
             if not output:
                 continue
-            if output[-1] == ')':
+            if output[-1] == ")":
                 # we have options
-                opt_begin = output.rfind('(')
-                options = output[opt_begin+1:-1]
+                opt_begin = output.rfind("(")
+                options = output[opt_begin + 1 : -1]
                 output = output[:opt_begin]
             else:
                 options = None
 
-            if output not in (C.LOG_OPT_OUTPUT_DEFAULT, C.LOG_OPT_OUTPUT_FILE, C.LOG_OPT_OUTPUT_MEMORY):
+            if output not in (
+                C.LOG_OPT_OUTPUT_DEFAULT,
+                C.LOG_OPT_OUTPUT_FILE,
+                C.LOG_OPT_OUTPUT_MEMORY,
+            ):
                 raise ValueError("Invalid output [%s]" % output)
 
             if output == C.LOG_OPT_OUTPUT_DEFAULT:
@@ -314,20 +335,26 @@
                 handlers[output] = None
             elif output == C.LOG_OPT_OUTPUT_FILE:
                 if not options:
-                    ValueError("{handler} output need a path as option" .format(handle=output))
+                    ValueError(
+                        "{handler} output need a path as option".format(handle=output)
+                    )
                 handlers.setdefault(output, []).append(options)
-                options = None # option are parsed, we can empty them
+                options = None  # option are parsed, we can empty them
             elif output == C.LOG_OPT_OUTPUT_MEMORY:
                 # we have memory handler, option can be the len limit or None
                 try:
                     limit = int(options)
-                    options = None # option are parsed, we can empty them
+                    options = None  # option are parsed, we can empty them
                 except (TypeError, ValueError):
                     limit = C.LOG_OPT_OUTPUT_MEMORY_LIMIT
                 handlers[output] = limit
 
-            if options: # we should not have unparsed options
-                raise ValueError("options [{options}] are not supported for {handler} output".format(options=options, handler=output))
+            if options:  # we should not have unparsed options
+                raise ValueError(
+                    "options [{options}] are not supported for {handler} output".format(
+                        options=options, handler=output
+                    )
+                )
 
     @staticmethod
     def memory_get(size=None):
@@ -349,13 +376,12 @@
         try:
             start = cls.taints[level]
         except KeyError:
-            start = ''
+            start = ""
 
         if cls._color_location:
-            return message % {'color_start': start,
-                              'color_end': A.RESET}
+            return message % {"color_start": start, "color_end": A.RESET}
         else:
-            return '%s%s%s' % (start, message, A.RESET)
+            return "%s%s%s" % (start, message, A.RESET)
 
     @staticmethod
     def get_profile():
@@ -370,9 +396,10 @@
         ConfigureCustom.LOGGER_CLASS = logger_class
 
 
-configure_cls = { None: ConfigureBase,
-                   C.LOG_BACKEND_CUSTOM: ConfigureCustom
-                 }  # XXX: (key: backend, value: Configure subclass) must be filled when new backend are added
+configure_cls = {
+    None: ConfigureBase,
+    C.LOG_BACKEND_CUSTOM: ConfigureCustom,
+}  # XXX: (key: backend, value: Configure subclass) must be filled when new backend are added
 
 
 def configure(backend_, **options):
@@ -391,36 +418,46 @@
     except KeyError:
         raise ValueError("unknown backend [{}]".format(backend))
     if backend == C.LOG_BACKEND_CUSTOM:
-        logger_class = options.pop('logger_class')
+        logger_class = options.pop("logger_class")
         configure_class(logger_class, **options)
     else:
         configure_class(**options)
 
+
 def memory_get(size=None):
     if not C.LOG_OPT_OUTPUT_MEMORY in handlers:
-        raise ValueError('memory output is not used')
+        raise ValueError("memory output is not used")
     return configure_cls[backend].memory_get(size)
 
+
 def getLogger(name=C.LOG_BASE_LOGGER) -> Logger:
     try:
         logger_class = configure_cls[backend].LOGGER_CLASS
     except KeyError:
-        raise ValueError("This method should not be called with backend [{}]".format(backend))
+        raise ValueError(
+            "This method should not be called with backend [{}]".format(backend)
+        )
     return _loggers.setdefault(name, logger_class(name))
 
+
 _root_logger = getLogger()
 
+
 def debug(msg, **kwargs):
     _root_logger.debug(msg, **kwargs)
 
+
 def info(msg, **kwargs):
     _root_logger.info(msg, **kwargs)
 
+
 def warning(msg, **kwargs):
     _root_logger.warning(msg, **kwargs)
 
+
 def error(msg, **kwargs):
     _root_logger.error(msg, **kwargs)
 
+
 def critical(msg, **kwargs):
     _root_logger.critical(msg, **kwargs)