# -*- coding: utf-8 -*- import abc import json import logging import logging.config from typing import Any, Dict, Optional, Tuple # pylint: disable=W0611 LOG = logging.getLogger(__name__) class LogData(metaclass=abc.ABCMeta): """Logging data""" FIELDS = () # type: Tuple[str, ...] RAW_FIELDS = () # type: Tuple[str, ...] def __init__(self, record: Optional[logging.LogRecord] = None) -> None: self.data = {} # type: Dict[str, Any] self.freefields = False if not hasattr(self, 'freefields') else self.freefields # type: bool self.record = None # type: Optional[logging.LogRecord] if isinstance(record, logging.LogRecord): self.load(record) def get(self, field: Optional[str] = None) -> Any: """Get the data from a single field. :param field: The field to get """ if field is None: # Return 'data' dict return self.data if field in self.data: # Return particular field from 'data' dict return self.data[field] return None def getDict(self) -> Dict[str, Any]: """Get a dictionary of the data.""" datadict = {} for key, val in self.get().items(): if isinstance(val, LogData): datadict[key] = val.getDict() else: datadict[key] = val return datadict def getJSON(self) -> str: return json.dumps(self.getDict()) def parseVal(self, field: str, value: Any) -> Any: """Parse a value and coerce it into appropriate datatypes. :param field: The name of the field to be parsed :param value: The value of the field. """ if isinstance(value, dict): for k, v in value.items(): value[k] = self.parseVal(k, v) return value # process_time needs to be a float for Kibana analysis if (field in self.RAW_FIELDS or value is None or field == "process_time" or isinstance(value, (LogData, bool))): return value # If we haven't specified otherwise, make sure value is a string, # for consistency of data type and to make readable in logs/ES/Kibana return str(value) def set(self, field: str, value: Any): """Set a value in the data dictionary. :param field: The field to be set :param value: The value of the field :raises ValueError: If the field name is invalid """ if field in self.FIELDS or self.freefields: self.data[field] = self.parseVal(field, value) else: raise ValueError("No such field '" + field + "'") return self def load(self, record: logging.LogRecord) -> None: """Load a log record. :param record: The record to be loaded """ self.record = record # Child should implement any class-specific loading code