Skip to content

Commit

Permalink
Refactor code to lower cohesion
Browse files Browse the repository at this point in the history
  • Loading branch information
Michal Ploski committed Dec 24, 2021
1 parent bc29018 commit 8059b58
Showing 1 changed file with 28 additions and 28 deletions.
56 changes: 28 additions & 28 deletions aws_lambda_powertools/logging/utils.py
Original file line number Diff line number Diff line change
@@ -1,57 +1,57 @@
import logging
from typing import List, Optional, TypeVar
from typing import Callable, List, Optional, TypeVar

from .logger import Logger

PowertoolsLogger = TypeVar("PowertoolsLogger", bound=Logger)


def copy_config_to_registered_loggers(
source_logger: PowertoolsLogger, exclude: Optional[List[str]] = None, include: Optional[List[str]] = None
source_logger: PowertoolsLogger,
exclude: Optional[List[str]] = None,
include: Optional[List[str]] = None,
) -> None:
"""Enable powertools logging for imported libraries.
Attach source logger handlers to external loggers.
Modify logger level based on source logger attribute.
Ensure powertools logger itself is excluded from registered list.
"""
registered_loggers = _find_registered_loggers(source_logger, exclude, include)

if include and not exclude:
loggers = include
filter_func = _include_registered_loggers_filter
elif include and exclude:
exclude = [source_logger.name, *exclude]
loggers = list(set(include) - set(exclude))
filter_func = _include_registered_loggers_filter
elif not include and exclude:
loggers = [source_logger.name, *exclude]
filter_func = _exclude_registered_loggers_filter
else:
loggers = [source_logger.name]
filter_func = _exclude_registered_loggers_filter

registered_loggers = _find_registered_loggers(source_logger, loggers, filter_func)
for logger in registered_loggers:
_configure_logger(source_logger, logger)


def _include_registered_loggers_filter(logger_list: List[str]):
return [
logging.getLogger(name) for name in logging.root.manager.loggerDict if "." not in name and name in logger_list
]
def _include_registered_loggers_filter(loggers: List[str]):
return [logging.getLogger(name) for name in logging.root.manager.loggerDict if "." not in name and name in loggers]


def _exclude_registered_loggers_filter(logger_list: List[str]) -> List[logging.Logger]:
def _exclude_registered_loggers_filter(loggers: List[str]) -> List[logging.Logger]:
return [
logging.getLogger(name)
for name in logging.root.manager.loggerDict
if "." not in name and name not in logger_list
logging.getLogger(name) for name in logging.root.manager.loggerDict if "." not in name and name not in loggers
]


def _find_registered_loggers(
source_logger: PowertoolsLogger, exclude: Optional[List[str]] = None, include: Optional[List[str]] = None
source_logger: PowertoolsLogger, loggers: List[str], filter_func: Callable
) -> List[logging.Logger]:
"""Filter root loggers based on provided parameters.
Ensure powertools logger itself is excluded from final list.
"""
root_loggers = []
if include and not exclude:
root_loggers = _include_registered_loggers_filter(logger_list=include)
elif include and exclude:
exclude = [source_logger.name, *exclude]
root_loggers = _include_registered_loggers_filter(logger_list=list(set(include) - set(exclude)))
elif not include and exclude:
exclude = [source_logger.name, *exclude]
root_loggers = _exclude_registered_loggers_filter(logger_list=exclude)
else:
root_loggers = _exclude_registered_loggers_filter(logger_list=[source_logger.name])

"""Filter root loggers based on provided parameters."""
root_loggers = filter_func(loggers)
source_logger.debug(f"Filtered root loggers: {root_loggers}")
return root_loggers

Expand Down

0 comments on commit 8059b58

Please sign in to comment.