Source code for jina.logging.formatter

import json
import logging
import re
from copy import copy
from logging import Formatter, LogRecord

from ..enums import LogVerbosity
from ..helper import colored


[docs]class ColorFormatter(Formatter): """Format the log into colored logs based on the log-level.""" MAPPING = { LogVerbosity.DEBUG: dict(color='magenta'), LogVerbosity.INFO: dict(), # plain LogVerbosity.SUCCESS: dict(color='green'), LogVerbosity.WARNING: dict(color='yellow'), LogVerbosity.ERROR: dict(color='red'), LogVerbosity.CRITICAL: dict(color='red', attrs=['bold']), } #: log-level to color mapping def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.formatters = { k: logging.Formatter(colored(self._fmt, **v)) for k, v in self.MAPPING.items() }
[docs] def format(self, record): """ Format the LogRecord with corresponding colour. :param record: A LogRecord object :return:: Formatted LogRecord with level-colour MAPPING to add corresponding colour. """ formatter = self.formatters.get(record.levelno) return formatter.format(record)
[docs]class PlainFormatter(Formatter): """Remove all control chars from the log and format it as plain text, also restrict the max-length of msg to 512."""
[docs] def format(self, record): """ Format the LogRecord by removing all control chars and plain text, and restrict the max-length of msg to 512. :param record: A LogRecord object. :return:: Formatted plain LogRecord. """ cr = copy(record) if isinstance(cr.msg, str): cr.msg = re.sub(r'\u001b\[.*?[@-~]', '', str(cr.msg))[:512] return super().format(cr)
[docs]class JsonFormatter(Formatter): """Format the log message as a JSON object so that it can be later used/parsed in browser with javascript.""" KEYS = { 'created', 'filename', 'funcName', 'levelname', 'lineno', 'msg', 'module', 'name', 'pathname', 'process', 'thread', 'processName', 'threadName', 'log_id', } #: keys to extract from the log
[docs] def format(self, record: 'LogRecord'): """ Format the log message as a JSON object. :param record: A LogRecord object. :return:: LogRecord with JSON format. """ cr = copy(record) cr.msg = re.sub(r'\u001b\[.*?[@-~]', '', str(cr.msg)) return json.dumps( {k: getattr(cr, k) for k in self.KEYS if hasattr(cr, k)}, sort_keys=True )
[docs]class ProfileFormatter(Formatter): """Format the log message as JSON object and add the current used memory into it."""
[docs] def format(self, record: 'LogRecord'): """ Format the log message as JSON object and add the current used memory. :param record: A LogRecord object. :return:: Return JSON formatted log if msg of LogRecord is dict type else return empty. """ from .profile import used_memory cr = copy(record) if isinstance(cr.msg, dict): cr.msg.update( {k: getattr(cr, k) for k in ['created', 'module', 'process', 'thread']} ) cr.msg['memory'] = used_memory(unit=1) return json.dumps(cr.msg, sort_keys=True) else: return ''