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)