This document describes the current stable version of Kombu (5.0). For development docs, go here.
Source code for kombu.utils.debug
"""Debugging support."""
import logging
from vine.utils import wraps
from kombu.log import get_logger
__all__ = ('setup_logging', 'Logwrapped')
[docs]def setup_logging(loglevel=logging.DEBUG, loggers=None):
"""Setup logging to stdout."""
loggers = ['kombu.connection', 'kombu.channel'] if not loggers else loggers
for logger_name in loggers:
logger = get_logger(logger_name)
logger.addHandler(logging.StreamHandler())
logger.setLevel(loglevel)
[docs]class Logwrapped:
"""Wrap all object methods, to log on call."""
__ignore = ('__enter__', '__exit__')
def __init__(self, instance, logger=None, ident=None):
self.instance = instance
self.logger = get_logger(logger)
self.ident = ident
def __getattr__(self, key):
meth = getattr(self.instance, key)
if not callable(meth) or key in self.__ignore:
return meth
@wraps(meth)
def __wrapped(*args, **kwargs):
info = ''
if self.ident:
info += self.ident.format(self.instance)
info += f'{meth.__name__}('
if args:
info += ', '.join(map(repr, args))
if kwargs:
if args:
info += ', '
info += ', '.join(f'{key}={value!r}'
for key, value in kwargs.items())
info += ')'
self.logger.debug(info)
return meth(*args, **kwargs)
return __wrapped
def __repr__(self):
return repr(self.instance)
def __dir__(self):
return dir(self.instance)