mypy: Add type hints to pglookout [BF-1560]

Makefile

@@ -3,7 +3,7 @@ long_ver = $(shell git describe --long 2>/dev/null || echo $(short_ver)-0-unknow
 generated = pglookout/version.py
 
 PYTHON ?= python3
-PYTHON_SOURCE_DIRS = pglookout/ test/
+PYTHON_SOURCE_DIRS = pglookout/ test/ stubs/
 
 all: $(generated)
 	: 'try "make rpm" or "make deb" or "make test"'
@@ -17,7 +17,7 @@ unittest: $(generated)
 	$(PYTHON) -m pytest
 
 mypy: $(generated)
-	$(PYTHON) -m mypy
+	MYPYPATH=stubs $(PYTHON) -m mypy
 
 flake8: $(generated)
 	$(PYTHON) -m flake8 $(PYTHON_SOURCE_DIRS)

pglookout/logutil.py

@@ -5,39 +5,50 @@
 Copyright (c) 2015 Ohmu Ltd
 See LICENSE for details
 """
+from __future__ import annotations
+
+from logging.handlers import SysLogHandler
+from typing import Final, TYPE_CHECKING
 
 import logging
-import logging.handlers
 import os
 
+with_systemd: bool = False
 try:
     from systemd import daemon  # pylint: disable=no-name-in-module
+
+    with_systemd = True
 except ImportError:
-    daemon = None
+    if not TYPE_CHECKING:
+        daemon = None
 
-LOG_FORMAT = "%(asctime)s\t%(name)s\t%(threadName)s\t%(levelname)s\t%(message)s"
-LOG_FORMAT_SHORT = "%(levelname)s\t%(message)s"
-LOG_FORMAT_SYSLOG = "%(name)s %(threadName)s %(levelname)s: %(message)s"
+LOG_FORMAT: Final[str] = "%(asctime)s\t%(name)s\t%(threadName)s\t%(levelname)s\t%(message)s"
+LOG_FORMAT_SHORT: Final[str] = "%(levelname)s\t%(message)s"
+LOG_FORMAT_SYSLOG: Final[str] = "%(name)s %(threadName)s %(levelname)s: %(message)s"
 
 
-def set_syslog_handler(address, facility, logger):
-    syslog_handler = logging.handlers.SysLogHandler(address=address, facility=facility)
+def set_syslog_handler(address: str, facility: str | int, logger: logging.Logger) -> SysLogHandler:
+    if isinstance(facility, str):
+        facility_id: int = SysLogHandler.facility_names.get(facility, SysLogHandler.LOG_LOCAL2)
+    else:
+        facility_id = facility
+    syslog_handler = SysLogHandler(address=address, facility=facility_id)
     logger.addHandler(syslog_handler)
     formatter = logging.Formatter(LOG_FORMAT_SYSLOG)
     syslog_handler.setFormatter(formatter)
     return syslog_handler
 
 
-def configure_logging(level=logging.DEBUG, short_log=False):
+def configure_logging(level: int = logging.DEBUG, short_log: bool = False) -> None:
     # Are we running under systemd?
     if os.getenv("NOTIFY_SOCKET"):
         logging.basicConfig(level=level, format=LOG_FORMAT_SYSLOG)
-        if not daemon:
+        if not with_systemd:
             print("WARNING: Running under systemd but python-systemd not available, systemd won't see our notifications")
     else:
         logging.basicConfig(level=level, format=LOG_FORMAT_SHORT if short_log else LOG_FORMAT)
 
 
-def notify_systemd(status):
-    if daemon:
+def notify_systemd(status: str) -> None:
+    if with_systemd:
         daemon.notify(status)

pyproject.toml

@@ -34,7 +34,6 @@ exclude = [
     'pglookout/cluster_monitor.py',
     'pglookout/common.py',
     'pglookout/current_master.py',
-    'pglookout/logutil.py',
     'pglookout/pglookout.py',
     'pglookout/pgutil.py',
     'pglookout/statsd.py',

stubs/__init__.pyi

@@ -0,0 +1 @@
+# Copyright (c) 2023 Aiven, Helsinki, Finland. https://aiven.io/

stubs/systemd/__init__.pyi

@@ -0,0 +1 @@
+# Copyright (c) 2023 Aiven, Helsinki, Finland. https://aiven.io/

stubs/systemd/daemon.pyi

@@ -0,0 +1,19 @@
+# Copyright (c) 2023 Aiven, Helsinki, Finland. https://aiven.io/
+
+from __future__ import annotations
+
+from typing import AnyStr, IO
+
+def notify(
+    status: str,
+    unset_environment: bool = False,
+    pid: int = 0,
+    fds: IO[AnyStr] | None = None,
+) -> bool:
+    """
+    notify(status, unset_environment=False, pid=0, fds=None) -> bool
+
+    Send a message to the init system about a status change.
+    Wraps sd_notify(3).
+    """
+    ...