Logging in Python using structlog and canonicaljson

default_config.yaml

# file: default_config.yaml

---
version: 1
disable_existing_loggers: False
formatters:
  simple:
    format: "%(message)s"

filters:
  stderr_filter:
    (): setup_logger.StdErrFilter
  stdout_filter:
    (): setup_logger.StdOutFilter

handlers:
  filehandler:
    class: logging.handlers.WatchedFileHandler
    level: INFO
    formatter: simple
    filename: app.log
  
  stdout:
    class: logging.StreamHandler
    level: DEBUG
    formatter: simple
    filters: [stdout_filter]
    stream: ext://sys.stdout

  stderr:
    class: logging.StreamHandler
    level: ERROR
    formatter: simple
    filters: [stderr_filter]
    stream: ext://sys.stderr

loggers:
  app:
    level: INFO
    handlers: [stdout, stderr, filehandler]
    propagate: no

root:
  level: DEBUG
  handlers: [stdout, stderr, filehandler]

file.py

# file: file.py

from logger import logger

logger.info(event="Hello World!", user="Abhishek")

logger.py

# file: logger.py

import logging

import structlog

from setup_logger import setup_logging


logging.getLogger('requests').setLevel(logging.CRITICAL)
logging.getLogger('urllib3').setLevel(logging.CRITICAL)

setup_logging(thread_local=True)

logger = structlog.get_logger()

requirements.txt

canonicaljson==1.1.4
PyYAML==5.1.2
structlog==19.1.0

setup_logger.py

# file: setup_logger.py
# -*- coding: utf-8 -*-

"""Main module."""

import os
import datetime
import logging
import logging.config

import yaml
import structlog
import canonicaljson


DEFAULT_LOG_CONFIG_YAML = 'default_config.yaml'


class StdLibBoundLogger(structlog.stdlib.BoundLogger):
    """
    Adds msg method on `structlog.stdlib.BoundLogger` to be compatible with
    other modules.
    """
    def msg(self, event, *args, **kw):
        """
        Process event and call the appropriate logging method depending on
        `level`.
        """
        level = kw.get('level', 'debug')
        return self._proxy_to_logger(level, event, *args, **kw)


def setup_logging(thread_local=False) -> None:
    """
    Setup logging configuration

    Args:
        thread_local: Boolean to configure structlog to use threadlocal context
            See https://www.structlog.org/en/stable/thread-local.html#wrapped-dicts
    """
    
    default_config_path = os.path.join(
        os.path.dirname(os.path.abspath(__file__)),
        DEFAULT_LOG_CONFIG_YAML
    )
    if not os.path.exists(default_config_path):
        _print_with_ts('Pre-defined log config not found, exiting application')
        _exit_application()

    with open(default_config_path, 'rt') as f:
        config = yaml.safe_load(f.read())

    logging.config.dictConfig(config)

    # Set the context_class to be used for configuring structlog
    context_class = None
    if thread_local:
        context_class = structlog.threadlocal.wrap_dict(dict)

    # Configure Structlog wrapper for client use
    structlog.configure(
        processors=[
            structlog.stdlib.filter_by_level,
            structlog.stdlib.add_logger_name,
            structlog.stdlib.add_log_level,
            structlog.stdlib.PositionalArgumentsFormatter(),
            structlog.processors.TimeStamper(
                fmt='iso',
                utc=True,
                key='timestamp_logged'
            ),
            structlog.processors.StackInfoRenderer(),
            structlog.processors.format_exc_info,
            structlog.processors.UnicodeDecoder(),
            canonical_json_renderer,
        ],
        context_class=context_class,
        logger_factory=structlog.stdlib.LoggerFactory(),
        wrapper_class=StdLibBoundLogger,
        cache_logger_on_first_use=True
    )

    # Log successful event initialization
    # If there's any issue with logging, the application will immediately exit
    logger = structlog.get_logger(__name__)
    logger.info(type='LOGGER_SETUP',
        event='Logger successfully initialized')


def _print_with_ts(msg):
    ts = datetime.datetime.utcnow().isoformat()
    print('{} - {}'.format(ts, msg))


def _exit_application():
    # Using sys.exit doesn't work when running flask server
    os._exit(os.EX_CONFIG)


class StdErrFilter(logging.Filter):
    """
    Filter for the stderr stream
    Doesn't print records below ERROR to stderr to avoid dupes
    """
    def filter(self, record):
        return 0 if record.levelno < logging.ERROR else 1


class StdOutFilter(logging.Filter):
    """
    Filter for the stdout stream
    Doesn't print records above WARNING to stdout to avoid dupes
    """
    def filter(self, record):
        return 1 if record.levelno < logging.ERROR else 0


def canonical_json_renderer(logger, log_method, event_dict):
    """Wrapper around canonicaljson.encode_canonical_json that decodes json
    to str.
    """
    return canonicaljson.encode_canonical_json(event_dict).decode('utf-8')