Skip to content

Instantly share code, notes, and snippets.

@samkeen
Forked from impredicative/logger.py
Created March 13, 2020 22:02
Show Gist options
  • Save samkeen/c8f65866edc96fd8d5feaabd256307c0 to your computer and use it in GitHub Desktop.
Save samkeen/c8f65866edc96fd8d5feaabd256307c0 to your computer and use it in GitHub Desktop.
Python json logging using structlog and stdlib logging
"""
Usage example:
from logger import get_logger
log = get_logger()
log.info('my_event', my_key1='val 1', my_key2=5, my_key3=[1, 2, 3], my_key4={'a': 1, 'b': 2})
List of metadata keys in each log message:
event
_func
_level
_levelno
_lineno
_module
_thread_id
_thread_name
_time_unix
_time_human_localized
_time_human_utc
Requirements: pytz
Limitations: multithreading is supported but not multiprocessing.
"""
import collections
import datetime
import inspect
import logging
import logging.config
import os
import platform
import sys
import tempfile
import threading
import time
import structlog
from myapp.config import PACKAGE_NAME
import pytz
BASE_LOGGER_NAME = PACKAGE_NAME
IS_CONFIGURED = False
LOGGED_TZ = pytz.timezone('US/Eastern')
TEMPDIR = '/tmp' if platform.system() == 'Darwin' else tempfile.gettempdir()
LOGGING_CONFIG = {
'version': 1,
'disable_existing_loggers': False,
'handlers': {
'file': {
'()': logging.handlers.TimedRotatingFileHandler,
'filename': os.path.join(os.getenv('LOGDIR', TEMPDIR), BASE_LOGGER_NAME + '.log'),
'when': 'midnight',
'backupCount': 14,
'utc': True,
},
},
'loggers': {
BASE_LOGGER_NAME: {
'propagate': False,
'handlers': ['file'],
'level': 'DEBUG',
},
},
}
def _event_uppercase(logger, method_name, event_dict): # pylint: disable=unused-argument
event_dict['event'] = event_dict['event'].upper()
return event_dict
def _add_timestamp(logger, method_name, event_dict): # pylint: disable=unused-argument
event_dict['_time_unix'] = time.time()
dt_utc = datetime.datetime.fromtimestamp(event_dict['_time_unix'], datetime.timezone.utc)
event_dict['_time_human_utc'] = dt_utc.isoformat()
event_dict['_time_human_localized'] = {LOGGED_TZ.zone: dt_utc.astimezone(LOGGED_TZ).isoformat()}
return event_dict
def _add_caller_info(logger, method_name, event_dict): # pylint: disable=unused-argument
# Typically skipped funcs: _add_caller_info, _process_event, _proxy_to_logger, _proxy_to_logger
frame = inspect.currentframe()
while frame:
frame = frame.f_back
module = frame.f_globals['__name__']
if module.startswith('structlog.'):
continue
event_dict['_module'] = module
event_dict['_lineno'] = frame.f_lineno
event_dict['_func'] = frame.f_code.co_name
return event_dict
def _add_log_level(logger, method_name, event_dict): # pylint: disable=unused-argument
event_dict['_level'] = method_name
event_dict['_levelno'] = getattr(logging, method_name.upper())
return event_dict
def _add_thread_info(logger, method_name, event_dict): # pylint: disable=unused-argument
thread = threading.current_thread()
event_dict['_thread_id'] = thread.ident
event_dict['_thread_name'] = thread.name
return event_dict
def _order_keys(logger, method_name, event_dict): # pylint: disable=unused-argument
return collections.OrderedDict(sorted(event_dict.items(), key=lambda item: (item[0] != 'event', item)))
def _setup_once():
structlog.configure_once(
processors=[
structlog.stdlib.filter_by_level,
_add_caller_info,
_add_log_level,
_add_thread_info,
_event_uppercase,
# structlog.stdlib.add_logger_name, # Typically a duplicate of "_module"
structlog.stdlib.PositionalArgumentsFormatter(True),
_add_timestamp,
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
structlog.processors.UnicodeEncoder() if sys.version_info.major == 2 else
structlog.processors.UnicodeDecoder(),
_order_keys,
structlog.processors.JSONRenderer(),
],
wrapper_class=structlog.stdlib.BoundLogger,
context_class=dict,
logger_factory=structlog.stdlib.LoggerFactory(),
cache_logger_on_first_use=True,
)
logging.config.dictConfig(LOGGING_CONFIG)
logger = get_logger(__name__)
logger.info('logging_initialized',
logger_level=LOGGING_CONFIG['loggers'][BASE_LOGGER_NAME]['level'].lower(),
logfile=LOGGING_CONFIG['handlers']['file']['filename'],
)
def get_logger(logger_name=None):
global IS_CONFIGURED # pylint: disable=global-statement
if not IS_CONFIGURED:
IS_CONFIGURED = True
_setup_once()
if logger_name is None:
logger_name = inspect.currentframe().f_back.f_globals['__name__']
logger_name = BASE_LOGGER_NAME if logger_name == '__main__' else logger_name
return structlog.wrap_logger(logging.getLogger(logger_name))
if __name__ == '__main__':
log = get_logger()
log.debug('test')
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment