I usually use Python mainly, but structural data is more convenient when developing a log collection / analysis platform with a cloud service, isn't it? However, if you use Python's standard library Logging's normal Formatter, you have to convert it to structural data once. It's unreasonable to create another application just for conversion. Therefore, I developed a Formatter that outputs logs in JSON.
--OS (guaranteed operation) - MacOS Catalina - Ubuntu 18.04 --Language - Python ^3.7
--Package manager - Poetry
--Development library - Black - Pytest - Flake8
The developed Formatter can be found at homoluctus / json-pyformatter. It is also published on PyPI-> json-pyformatter 0.1.0
pip install json-pyformatter
The field that can be output is logrecord-attributes of the standard library Logging.
The default fields output by the developed Formatter are ʻasctime, levelname, message`.
import logging
from json_pyformmatter import JsonFormatter
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
handler = logging.StreamHandler()
fields = ('levelname', 'filename', 'message')
formatter = JsonFormatter(fields=fields)
handler.setFormatter(formatter)
logger.addHandler(hander)
logger.info('hello')
When this is executed, the JSON log will be output as shown below.
{"levelname": "INFO", "filename": "test_formatter.py", "message": "hello"}
Also, it will be easier to see if you specify ʻindent = 2` in the argument of JsonFormatter.
{
"levelname": "INFO",
"filename": "test_formatter.py",
"message": "hello"
}
Of course, you can also output traceback. Since it is difficult to see if it is one line, it is arranged in an array.
{
'asctime': '2019-12-01 13:58:34',
'levelname': 'ERROR',
'message': 'error occurred !!',
'traceback': [
'Traceback (most rec...ll last):',
'File "/example/test..._exc_info',
'raise TypeError(message)',
'TypeError: error occurred !!'
]
}
From here, I will explain the source code.
I will write all the source code for the time being.
import json
from collections import OrderedDict
from logging import Formatter
class JsonFormatter(Formatter):
default_fields = ('asctime', 'levelname', 'message')
def __init__(self, fields=None, datefmt=None, indent=None):
"""
Args:
fields (tuple, list)
datefmt (str)
indent (str, int)
"""
self.fields = (
self.get_or_none(fields, (list, tuple)) or self.default_fields
)
# default time format is %Y-%m-%d %H:%M:%S
self.datefmt = (
self.get_or_none(datefmt, str) or self.default_time_format
)
self._indent = self.get_or_none(indent, (str, int))
def get_or_none(self, target, types):
"""Check whether target value is expected type.
If target type does not match expected type, returns None.
Args:
target (any)
types (class, tuple)
Returns:
target or None
"""
if isinstance(target, types):
return target
return None
def getMessage(self, record):
if isinstance(record.msg, (list, tuple, dict)):
return record.msg
return record.getMessage()
def _format_json(self, record):
return json.dumps(record, ensure_ascii=False, indent=self._indent)
def _format(self, record):
log = OrderedDict()
try:
for field in self.fields:
log[field] = getattr(record, field)
return log
except AttributeError as err:
raise ValueError(f'Formatting field not found in log record {err}')
def format(self, record):
record.message = self.getMessage(record)
record.asctime = self.formatTime(record, self.datefmt)
formatted_record = self._format(record)
if record.exc_info:
if not record.exc_text:
record.exc_text = self.formatException(record.exc_info)
if record.exc_text:
formatted_record['traceback'] = [
msg.strip() for msg in record.exc_text.strip().split('\n')
]
if record.stack_info:
formatted_record['stack'] = record.stack_info.strip()
return self._format_json(formatted_record)
get_or_none This method is used when creating an instance. If it is not the expected type, use the default value or use it to assign None as it is.
def get_or_none(self, target, types):
"""Check whether target value is expected type.
If target type does not match expected type, returns None.
Args:
target (any)
types (class, tuple)
Returns:
target or None
"""
if isinstance(target, types):
return target
return None
getMessage
It is a method to get the message of the argument of logger.info (). The return value of this getMessage is set in the JSON message field.
If the instance type of record.msg is any of (list, tuple, dict), it is returned as it is, otherwise, the return value of the getMessage method of the record instance is returned. By doing so, when any of (list, tuple, dict) is passed, the log can be output as a JSON array / object.
def getMessage(self, record):
if isinstance(record.msg, (list, tuple, dict)):
return record.msg
return record.getMessage()
_format_json This method dumps the argument record as JSON.
def _format_json(self, record):
return json.dumps(record, ensure_ascii=False, indent=self._indent)
_format
A method to convert record to a Python dictionary.
The field specified by the user does not necessarily exist in the attribute of record, so it is try-except. The log of the return value is set to ʻOrderedDict` to guarantee the order.
def _format(self, record):
log = OrderedDict()
try:
for field in self.fields:
log[field] = getattr(record, field)
return log
except AttributeError as err:
raise ValueError(f'Formatting field not found in log record {err}')
format This method is called from logging.Handler. I don't think you will create your own Formatter and use the format of the parent class (logging.Formatter), so be sure to overwrite this method. I'm formatting record.exc_text to output traceback as an array. After removing extra spaces etc., it is split with a line break to make an array. Finally, I dump the Python dict as JSON.
def format(self, record):
record.message = self.getMessage(record)
record.asctime = self.formatTime(record, self.datefmt)
formatted_record = self._format(record)
if record.exc_info:
if not record.exc_text:
record.exc_text = self.formatException(record.exc_info)
if record.exc_text:
formatted_record['traceback'] = [
msg.strip() for msg in record.exc_text.strip().split('\n')
]
if record.stack_info:
formatted_record['stack'] = record.stack_info.strip()
return self._format_json(formatted_record)
--logging --- logging function for Python
Recommended Posts