Merge pull request #68 from jd/more-typing

chore: type everything

daiquiri/__init__.py

@@ -15,12 +15,18 @@
 import logging.handlers
 import sys
 import traceback
+import types
 import typing
 
 from daiquiri import output
 
+if typing.TYPE_CHECKING:
+    _KeywordArgumentAdapterBase = logging.LoggerAdapter[logging.Logger]
+else:
+    _KeywordArgumentAdapterBase = logging.LoggerAdapter
 
-class KeywordArgumentAdapter(logging.LoggerAdapter):
+
+class KeywordArgumentAdapter(_KeywordArgumentAdapterBase):
     """Logger adapter to add keyword arguments to log record's extra data.
 
     Keywords passed to the log call are added to the "extra"
@@ -55,7 +61,9 @@ def process(
         return msg, kwargs
 
 
-def getLogger(name: typing.Optional[str] = None, **kwargs) -> KeywordArgumentAdapter:
+def getLogger(
+    name: typing.Optional[str] = None, **kwargs: typing.Any
+) -> KeywordArgumentAdapter:
     """Build a logger with the given name.
 
     :param name: The name for the logger. This is usually the module
@@ -67,11 +75,11 @@ def getLogger(name: typing.Optional[str] = None, **kwargs) -> KeywordArgumentAda
 
 def setup(
     level: int = logging.WARNING,
-    outputs: typing.List[output.Output] = [output.STDERR],
+    outputs: typing.Iterable[typing.Union[output.Output, str]] = [output.STDERR],
     program_name: typing.Optional[str] = None,
     capture_warnings: bool = True,
     set_excepthook: bool = True,
-):
+) -> None:
     """Set up Python logging.
 
     This sets up basic handlers for Python logging.
@@ -101,7 +109,11 @@ def setup(
     if set_excepthook:
         program_logger = logging.getLogger(program_name)
 
-        def logging_excepthook(exc_type, value, tb):
+        def logging_excepthook(
+            exc_type: typing.Optional[typing.Type[BaseException]],
+            value: typing.Optional[BaseException],
+            tb: typing.Optional[types.TracebackType],
+        ) -> None:
             program_logger.critical(
                 "".join(traceback.format_exception(exc_type, value, tb))
             )

daiquiri/formatter.py

@@ -12,6 +12,7 @@
 """Formatters."""
 
 import logging
+import typing
 
 from pythonjsonlogger import jsonlogger
 
@@ -41,21 +42,21 @@ class ColorFormatter(logging.Formatter):
 
     COLOR_STOP = "\033[0m"
 
-    def add_color(self, record):
+    def add_color(self, record: logging.LogRecord) -> None:
         """Add color to a record."""
         if getattr(record, "_stream_is_a_tty", False):
-            record.color = self.LEVEL_COLORS[record.levelno]
-            record.color_stop = self.COLOR_STOP
+            record.color = self.LEVEL_COLORS[record.levelno]  # type: ignore[attr-defined]
+            record.color_stop = self.COLOR_STOP  # type: ignore[attr-defined]
         else:
-            record.color = ""
-            record.color_stop = ""
+            record.color = ""  # type: ignore[attr-defined]
+            record.color_stop = ""  # type: ignore[attr-defined]
 
-    def remove_color(self, record):
+    def remove_color(self, record: logging.LogRecord) -> None:
         """Remove color from a record."""
-        del record.color
-        del record.color_stop
+        del record.color  # type: ignore[attr-defined]
+        del record.color_stop  # type: ignore[attr-defined]
 
-    def format(self, record):
+    def format(self, record: logging.LogRecord) -> str:
         """Format a record."""
         self.add_color(record)
         s = super(ColorFormatter, self).format(record)
@@ -102,39 +103,39 @@ class ExtrasFormatter(logging.Formatter):
 
     def __init__(
         self,
-        keywords=None,
-        extras_template="[{0}: {1}]",
-        extras_separator=" ",
-        extras_prefix=" ",
-        extras_suffix="",
-        *args,
-        **kwargs,
-    ):
+        keywords: typing.Optional[typing.Set[str]] = None,
+        extras_template: str = "[{0}: {1}]",
+        extras_separator: str = " ",
+        extras_prefix: str = " ",
+        extras_suffix: str = "",
+        *args: typing.Any,
+        **kwargs: typing.Any,
+    ) -> None:
         self.keywords = set() if keywords is None else keywords
         self.extras_template = extras_template
         self.extras_separator = extras_separator
         self.extras_prefix = extras_prefix
         self.extras_suffix = extras_suffix
         super(ExtrasFormatter, self).__init__(*args, **kwargs)
 
-    def add_extras(self, record):
+    def add_extras(self, record: logging.LogRecord) -> None:
         if not hasattr(record, "_daiquiri_extra_keys"):
-            record.extras = ""
+            record.extras = ""  # type: ignore[attr-defined]
             return
 
         extras = self.extras_separator.join(
             self.extras_template.format(k, getattr(record, k))
-            for k in record._daiquiri_extra_keys
+            for k in record._daiquiri_extra_keys  # type: ignore[attr-defined]
             if k != "_daiquiri_extra_keys" and k not in self.keywords
         )
         if extras != "":
             extras = self.extras_prefix + extras + self.extras_suffix
-        record.extras = extras
+        record.extras = extras  # type: ignore[attr-defined]
 
-    def remove_extras(self, record):
-        del record.extras
+    def remove_extras(self, record: logging.LogRecord) -> None:
+        del record.extras  # type: ignore[attr-defined]
 
-    def format(self, record):
+    def format(self, record: logging.LogRecord) -> str:
         self.add_extras(record)
         s = super(ExtrasFormatter, self).format(record)
         self.remove_extras(record)
@@ -144,24 +145,29 @@ def format(self, record):
 class ColorExtrasFormatter(ColorFormatter, ExtrasFormatter):
     """Combines functionality of ColorFormatter and ExtrasFormatter."""
 
-    def format(self, record):
+    def format(self, record: logging.LogRecord) -> str:
         self.add_color(record)
         s = ExtrasFormatter.format(self, record)
         self.remove_color(record)
         return s
 
 
-class DatadogFormatter(jsonlogger.JsonFormatter):
-    def __init__(self):
+class DatadogFormatter(jsonlogger.JsonFormatter):  # type: ignore[misc]
+    def __init__(self) -> None:
         super(DatadogFormatter, self).__init__(timestamp=True)
 
-    def add_fields(self, log_record, record, message_dict):
+    def add_fields(
+        self,
+        log_record: typing.Dict[str, typing.Any],
+        record: logging.LogRecord,
+        message_dict: typing.Dict[str, str],
+    ) -> None:
         super(DatadogFormatter, self).add_fields(log_record, record, message_dict)
         log_record["status"] = record.levelname.lower()
         log_record["logger"] = {
             "name": record.name,
         }
-        if record.exc_info:
+        if record.exc_info and record.exc_info[0]:
             log_record["error"] = {
                 "kind": record.exc_info[0].__name__,
                 "stack": message_dict.get("stack_info"),

daiquiri/handlers.py

@@ -14,12 +14,14 @@
 import logging
 import logging.config
 import logging.handlers
+import typing
 
 try:
     from systemd import journal
 except ImportError:
     journal = None
 
+
 try:
     import syslog
 except ImportError:
@@ -43,7 +45,9 @@
 class SyslogHandler(logging.Handler):
     """Syslog based handler. Only available on UNIX-like platforms."""
 
-    def __init__(self, program_name, facility=None):
+    def __init__(
+        self, program_name: str, facility: typing.Optional[int] = None
+    ) -> None:
         # Default values always get evaluated, for which reason we avoid
         # using 'syslog' directly, which may not be available.
         facility = facility if facility is not None else syslog.LOG_USER
@@ -52,7 +56,7 @@ def __init__(self, program_name, facility=None):
         super(SyslogHandler, self).__init__()
         syslog.openlog(program_name, 0, facility)
 
-    def emit(self, record):
+    def emit(self, record: logging.LogRecord) -> None:
         priority = SYSLOG_MAP.get(record.levelname, 7)
         message = self.format(record)
         syslog.syslog(priority, message)
@@ -61,13 +65,15 @@ def emit(self, record):
 class JournalHandler(logging.Handler):
     """Journald based handler. Only available on platforms using systemd."""
 
-    def __init__(self, program_name):
+    program_name: str
+
+    def __init__(self, program_name: str) -> None:
         if not journal:
             raise RuntimeError("Systemd bindings do not exist")
         super(JournalHandler, self).__init__()
         self.program_name = program_name
 
-    def emit(self, record):
+    def emit(self, record: logging.LogRecord) -> None:
         priority = SYSLOG_MAP.get(record.levelname, 7)
         message = self.format(record)
 
@@ -90,47 +96,53 @@ def emit(self, record):
             extras["EXCEPTION_INFO"] = record.exc_info
 
         if hasattr(record, "_daiquiri_extra_keys"):
-            for k in record._daiquiri_extra_keys:
+            for k in record._daiquiri_extra_keys:  # type: ignore[attr-defined]
                 if k != "_daiquiri_extra_keys":
                     extras[k.upper()] = getattr(record, k)
 
         journal.send(message, **extras)
 
 
-class TTYDetectorStreamHandler(logging.StreamHandler):
+if typing.TYPE_CHECKING:
+    _TTYDetectorStreamHandlerBase = logging.StreamHandler[typing.Any]
+else:
+    _TTYDetectorStreamHandlerBase = logging.StreamHandler
+
+
+class TTYDetectorStreamHandler(_TTYDetectorStreamHandlerBase):
     """Stream handler that adds a hint in the record if the stream is a TTY."""
 
-    def format(self, record):
+    def format(self, record: logging.LogRecord) -> str:
         if hasattr(self.stream, "isatty"):
             try:
-                record._stream_is_a_tty = self.stream.isatty()
+                record._stream_is_a_tty = self.stream.isatty()  # type: ignore[attr-defined]
             except ValueError:
                 # Stream has been closed, usually during interpretor shutdown
-                record._stream_is_a_tty = False
+                record._stream_is_a_tty = False  # type: ignore[attr-defined]
         else:
-            record._stream_is_a_tty = False
+            record._stream_is_a_tty = False  # type: ignore[attr-defined]
         s = super(TTYDetectorStreamHandler, self).format(record)
-        del record._stream_is_a_tty
+        del record._stream_is_a_tty  # type: ignore[attr-defined]
         return s
 
 
 class PlainTextSocketHandler(logging.handlers.SocketHandler):
     """Socket handler that uses format and encode the record."""
 
-    def __init__(self, hostname, port, encoding="utf-8"):
+    def __init__(self, hostname: str, port: int, encoding: str = "utf-8") -> None:
         self.encoding = encoding
         super(PlainTextSocketHandler, self).__init__(hostname, port)
 
-    def makePickle(self, record):
+    def makePickle(self, record: logging.LogRecord) -> bytes:
         return self.format(record).encode(self.encoding) + b"\n"
 
 
 class PlainTextDatagramHandler(logging.handlers.DatagramHandler):
     """Socket handler that uses format and encode the record."""
 
-    def __init__(self, hostname, port, encoding="utf-8"):
+    def __init__(self, hostname: str, port: int, encoding: str = "utf-8") -> None:
         self.encoding = encoding
         super(PlainTextDatagramHandler, self).__init__(hostname, port)
 
-    def makePickle(self, record):
+    def makePickle(self, record: logging.LogRecord) -> bytes:
         return self.format(record).encode(self.encoding) + b"\n"