json_logging_logger.py

import logging
import os
import sys

import json_logging


class Logger:
    URGENT = logging.CRITICAL + 10

    def __init__(self, app_name, level=None):
        if not hasattr(logging, 'URGENT'):
            logging.addLevelName(self.URGENT, 'URGENT')
            logging.URGENT = self.URGENT

        if level is None:
            level = os.getenv('LOG_LEVEL', logging.INFO)

        self.logger = logging.getLogger(app_name)
        self.logger.setLevel(level)

        if not self.logger.handlers:
            handler = logging.StreamHandler(sys.stdout)
            if not json_logging.ENABLE_JSON_LOGGING:
                formatter = logging.Formatter('[%(asctime)s] [%(levelname)s] [%(name)s] %(message)s',
                                              datefmt='%Y-%m-%d %H:%M:%S')
                handler.setFormatter(formatter)
            self.logger.addHandler(handler)

    @staticmethod
    def init_json_logging(component, framework_name=None, app=None):
        """
        Usage:
            if config.JSON_LOGGING:
                Logger.init_json_logging('task_server', framework_name='flask', app=app)
        """
        json_logging.ENABLE_JSON_LOGGING = True
        json_logging.COMPONENT_NAME = component
        # json_logging.init(framework_name=framework_name)
        if app:
            json_logging.init_request_instrument(app)

    def set_level(self, level=logging.INFO):
        self.logger.setLevel(level)

    def _log(self, msg, args, props=None, level=logging.DEBUG):
        args_str = ' '.join(map(str, args))
        msg += args_str if args_str else ''

        if props:
            for k, v in props.items():
                props[k] = str(v)
            if not json_logging.ENABLE_JSON_LOGGING:
                props_str = ', '.join(f'{k}: {v}' for k, v in props.items())
                msg += '; ' + props_str

        self.logger.log(level, msg, extra={'props': props or {}})

    def log(self, msg, *args, level=logging.INFO, **props):
        self._log(msg, args, props, level=level)

    def __call__(self, msg, *args, **props):
        self._log(msg, args, props, level=logging.INFO)

    def urgent(self, msg, *args, **props):
        self.log(msg, args, props, level=self.URGENT)

    def critical(self, msg, *args, **props):
        self._log(msg, args, props, level=logging.CRITICAL)

    def error(self, msg, *args, **props):
        self._log(msg, args, props, level=logging.ERROR)

    def warning(self, msg, *args, **props):
        self._log(msg, args, props, level=logging.WARNING)

    def info(self, msg, *args, **props):
        self._log(msg, args, props, level=logging.INFO)

    def debug(self, msg, *args, **props):
        self._log(msg, args, props, level=logging.DEBUG)


def example_usage():
    Logger.init_json_logging('registry')

    log = Logger('registry_node_1')
    log.warning('warning')
    log.info('info here', a='a', b='b')


if __name__ == '__main__':
    example_usage()