Mercurial > libervia-backend
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)