This document is for Kombu's development version, which can be significantly different from previous releases. Get the stable docs here: 5.0.

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', ''] 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)