Created
March 23, 2021 15:40
-
-
Save MihanixA/81868444daa3de1a6204ee2e53d76d48 to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment