chore(typing): annotate utils/debug.py (#1714)

Co-authored-by: Stevie Gayet <stegayet@users.noreply.github.com>

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)

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)

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,