Python Logger Dict Formatter

logger_dict_formatter.py

from copy import deepcopy
from logging.config import dictConfig
from os.path import isfile

import yaml


class LoggerDictConfigFormatter:
    unwanted_keys: tuple = ('level', 'handlers', 'propagate')
    max_deep: int = 10
    input_dict: dict
    out_dict: dict

    def __init__(self, input_dict):
        self.input_dict = input_dict
        self.out_dict = self._format()

    def get_formatted(self):
        return self.out_dict

    def _format(self):
        if not self.input_dict.get('loggers'):
            return self.input_dict

        self.input_dict['loggers'] = self._format_loggers(self.input_dict['loggers'])
        return self.input_dict

    def _format_loggers(self, input_loggers, deep=1):
        if deep >= self.max_deep:
            return input_loggers
        if not self._has_children(input_loggers):
            return input_loggers

        out_loggers = deepcopy(input_loggers)
        for key in input_loggers:
            for subkey, subvalue in self._get_submodules(input_loggers[key]):
                out_loggers[f'{key}.{subkey}'] = out_loggers[key].pop(subkey)

        return self._format_loggers(self._remove_empty_values(out_loggers), deep + 1)

    def _has_children(self, input_loggers):
        return bool([it for it in input_loggers.items() if it[0] not in self.unwanted_keys])

    def _get_submodules(self, input_logger):
        return self._filter_submodules(input_logger).items()

    def _filter_submodules(self, input_logger):
        return {k: v for k, v in input_logger.items() if k not in self.unwanted_keys}

    def _remove_empty_values(self, the_dict):
        return {k: v for k, v in the_dict.items() if v}

test_logger_dict_formatter.py

from logging import getLogger, DEBUG, INFO, ERROR, NOTSET
from logging.config import dictConfig
from unittest import TestCase

from .logging import LoggerDictConfigFormatter


class TestLogging(TestCase):

    def test_log_submodule(self):
        dictConfig({
            'version': 1,
            'disable_existing_loggers': 'no',
            'handlers': {
                'console': {
                    'class': 'logging.StreamHandler',
                }
            },
            'loggers': {
                'root': {
                    'handlers': ['console'],
                },
                'package': {
                    'level': 'DEBUG',
                },
                'package.module': {
                    'level': 'INFO'
                },
                'package.module.module': {
                    'level': 'ERROR'
                }
            }
        })

        self.assertEqual(getLogger('package').level, DEBUG)

        self.assertEqual(getLogger('package.module').level, INFO)

        self.assertEqual(getLogger('package.module.module').level, ERROR)

        self.assertEqual(getLogger('package.anothermodule').level, NOTSET)

    def test_do_not_overwrite(self):
        dictConfig({
            'version': 1,
            'disable_existing_loggers': 'no',
            'handlers': {
                'console': {
                    'class': 'logging.StreamHandler',
                }
            },
            'loggers': {
                'root': {
                    'handlers': ['console'],
                },
                'package': {
                    'level': 'DEBUG',
                },
                'package.module': {
                    'level': 'INFO'
                },
                'package.module.module': {
                    'level': 'ERROR'
                }
            }
        })

        self.assertEqual(getLogger('package.anothermodule').level, NOTSET)

        dictConfig({
            'version': 1,
            'disable_existing_loggers': 'no',
            'loggers': {
                'package.anothermodule': {
                    'level': 'INFO'
                },
            }
        })

        self.assertEqual(getLogger('package').level, DEBUG)

        self.assertEqual(getLogger('package.module').level, INFO)

        self.assertEqual(getLogger('package.module.module').level, ERROR)

        self.assertEqual(getLogger('package.anothermodule').level, INFO)

    def test_dict_submodule(self):
        self.assertEqual(getLogger('package.anothermodule').level, NOTSET)
        given = {
            'version': 1,
            'disable_existing_loggers': 'no',
            'handlers': {
                'console': {
                    'class': 'logging.StreamHandler',
                }
            },
            'loggers': {
                'root_package': {
                    'level': 'DEBUG',
                    'submodule': {
                        'level': 'INFO'
                    },
                }
            }
        }
        _when = LoggerDictConfigFormatter(given).get_formatted()
        dictConfig(_when)

        self.assertEqual(getLogger('root_package').level, DEBUG)

        self.assertEqual(getLogger('root_package.submodule').level, INFO)

    def test_should_format_submodule(self):
        given = {
            'version': 1,
            'disable_existing_loggers': 'no',
            'handlers': {
                'console': {
                    'class': 'logging.StreamHandler',
                }
            },
            'loggers': {
                'root_package': {
                    'level': 'DEBUG',
                    'submodule': {
                        'level': 'INFO'
                    },
                }
            }
        }

        expected = {
            'version': 1,
            'disable_existing_loggers': 'no',
            'handlers': {
                'console': {
                    'class': 'logging.StreamHandler',
                }
            },
            'loggers': {
                'root_package': {
                    'level': 'DEBUG',
                },
                'root_package.submodule': {
                    'level': 'INFO'
                }
            }
        }

        self.assertEqual(expected, LoggerDictConfigFormatter(given).get_formatted())

    def test_should_format_submodule_two_levels(self):
        given = {
            'version': 1,
            'disable_existing_loggers': 'no',
            'handlers': {
                'console': {
                    'class': 'logging.StreamHandler',
                }
            },
            'loggers': {
                'root_package': {
                    'level': 'DEBUG',
                    'submodule': {
                        'anothersubmodule': {
                            'level': 'ERROR'
                        },
                    },
                }
            }
        }

        expected = {
            'version': 1,
            'disable_existing_loggers': 'no',
            'handlers': {
                'console': {
                    'class': 'logging.StreamHandler',
                }
            },
            'loggers': {
                'root_package': {
                    'level': 'DEBUG',
                },
                'root_package.submodule.anothersubmodule': {
                    'level': 'ERROR'
                }
            }
        }

        self.assertEqual(expected, LoggerDictConfigFormatter(given).get_formatted())