diff --git a/kombu/log.py b/kombu/log.py index ed8d0a50..b9bca7f1 100644 --- a/kombu/log.py +++ b/kombu/log.py @@ -7,11 +7,15 @@ import numbers import os import sys from logging.handlers import WatchedFileHandler +from typing import TYPE_CHECKING from .utils.encoding import safe_repr, safe_str from .utils.functional import maybe_evaluate from .utils.objects import cached_property +if TYPE_CHECKING: + from logging import Logger + __all__ = ('LogMixin', 'LOG_LEVELS', 'get_loglevel', 'setup_logging') LOG_LEVELS = dict(logging._nameToLevel) @@ -21,7 +25,7 @@ LOG_LEVELS.setdefault(logging.FATAL, 'FATAL') DISABLE_TRACEBACKS = os.environ.get('DISABLE_TRACEBACKS') -def get_logger(logger): +def get_logger(logger: str | Logger): """Get logger by name.""" if isinstance(logger, str): logger = logging.getLogger(logger) diff --git a/kombu/utils/debug.py b/kombu/utils/debug.py index bd20948f..7889ed47 100644 --- a/kombu/utils/debug.py +++ b/kombu/utils/debug.py @@ -3,15 +3,25 @@ from __future__ import annotations import logging +from typing import TYPE_CHECKING from vine.utils import wraps from kombu.log import get_logger +if TYPE_CHECKING: + from logging import Logger + from typing import Any, Callable, Dict, List, Optional + + from kombu.transport.base import Transport + __all__ = ('setup_logging', 'Logwrapped') -def setup_logging(loglevel=logging.DEBUG, loggers=None): +def setup_logging( + loglevel: Optional[int] = logging.DEBUG, + loggers: Optional[List[str]] = None +) -> None: """Setup logging to stdout.""" loggers = ['kombu.connection', 'kombu.channel'] if not loggers else loggers for logger_name in loggers: @@ -25,19 +35,24 @@ class Logwrapped: __ignore = ('__enter__', '__exit__') - def __init__(self, instance, logger=None, ident=None): + def __init__( + self, + instance: Transport, + logger: Optional[Logger] = None, + ident: Optional[str] = None + ): self.instance = instance self.logger = get_logger(logger) self.ident = ident - def __getattr__(self, key): + def __getattr__(self, key: str) -> Callable: meth = getattr(self.instance, key) if not callable(meth) or key in self.__ignore: return meth @wraps(meth) - def __wrapped(*args, **kwargs): + def __wrapped(*args: List[Any], **kwargs: Dict[str, Any]) -> Callable: info = '' if self.ident: info += self.ident.format(self.instance) @@ -55,8 +70,8 @@ class Logwrapped: return __wrapped - def __repr__(self): + def __repr__(self) -> str: return repr(self.instance) - def __dir__(self): + def __dir__(self) -> List[str]: return dir(self.instance) diff --git a/setup.cfg b/setup.cfg index c087af81..ced53758 100644 --- a/setup.cfg +++ b/setup.cfg @@ -25,6 +25,7 @@ disallow_untyped_defs = True ignore_missing_imports = True files = kombu/abstract.py, + kombu/utils/debug.py, kombu/utils/time.py, kombu/utils/uuid.py, t/unit/utils/test_uuid.py,