From 49160e6a5fcfcd9d2dd3511451d1ae68d59d3aed Mon Sep 17 00:00:00 2001
From: Mynhardt Burger <mynhardt@gmail.com>
Date: Sun, 28 Jan 2024 16:20:47 -0500
Subject: [PATCH 1/7] Typing and fix spelling mistakes

Signed-off-by: Mynhardt Burger <mynhardt@gmail.com>
---
 src/python/alog/alog.py | 79 ++++++++++++++++++++++-------------------
 1 file changed, 42 insertions(+), 37 deletions(-)

diff --git a/src/python/alog/alog.py b/src/python/alog/alog.py
index 663bbe3..282e1ec 100644
--- a/src/python/alog/alog.py
+++ b/src/python/alog/alog.py
@@ -34,6 +34,9 @@
 import time
 import traceback
 from datetime import datetime, timedelta
+from typing import Any, Callable, Union, Optional
+
+_Level = Union[int, str]
 
 ## Formatters ##################################################################
 
@@ -66,7 +69,7 @@ def __init__(self):
         # Initialize the underlying logger with this formatter
         logging.Formatter.__init__(self)
 
-    def formatTime(self, record, datefmt=None):
+    def formatTime(self, record:logging.LogRecord, datefmt:Optional[str]=None):
         """A wrapper for the parent formatTime that returns UTC timezone
         time stamp inISO format.
 
@@ -101,7 +104,7 @@ def __init__(self):
         AlogFormatterBase.__init__(self)
 
     @staticmethod
-    def _map_to_common_key_name(log_record_keyname):
+    def _map_to_common_key_name(log_record_keyname:str):
         if log_record_keyname == 'levelname':
             return 'level'
         elif log_record_keyname == 'asctime':
@@ -113,7 +116,7 @@ def _map_to_common_key_name(log_record_keyname):
         else:
             return log_record_keyname
 
-    def _extract_fields_from_record_as_dict(self, record):
+    def _extract_fields_from_record_as_dict(self, record:logging.LogRecord):
         """Extracts the fields we want out of log record and puts them into an
         dict for easy jsonification.
 
@@ -124,7 +127,7 @@ def _extract_fields_from_record_as_dict(self, record):
             The relevant fields pulled out from the log record object and
             initialized into a dictionary.
         """
-        out = {}
+        out:dict[Any, Any] = {}
         for field_name in self._FIELDS_TO_PRINT:
             if hasattr(record, field_name):
                 record_field = getattr(record, field_name)
@@ -136,7 +139,7 @@ def _extract_fields_from_record_as_dict(self, record):
         out["level"] = out["level"].lower()
         return out
 
-    def format(self, record):
+    def format(self, record:logging.LogRecord):
         """Formats the log record as a JSON formatted string
        (also removes new line characters so everything prints on a single line)
 
@@ -148,7 +151,7 @@ def format(self, record):
         """
         # Maintain the message as a dict if passed in as one
         if isinstance(record.msg, dict):
-            record.message = record.msg
+            record.message = record.msg # type: ignore
         else:
             record.message = record.getMessage()
 
@@ -201,7 +204,7 @@ def __init__(self, channel_len=5):
         AlogFormatterBase.__init__(self)
         self.channel_len = channel_len
 
-    def _make_header(self, timestamp, channel, level, log_code):
+    def _make_header(self, timestamp:str, channel:str, level:str, log_code:Optional[str]):
         """Create the header for a log line with proper padding.
         """
         # Get the padded or truncated channel
@@ -227,7 +230,7 @@ def _make_header(self, timestamp, channel, level, log_code):
 
         return header
 
-    def format(self, record):
+    def format(self, record:logging.LogRecord):
         """Formats the log record as pretty-printed lines of the format:
 
         timestamp [CHANL:LEVL] message
@@ -258,7 +261,7 @@ def format(self, record):
         level = record.levelname
         channel = record.name
         timestamp = self.formatTime(record, self.datefmt)
-        log_code = record.log_code if hasattr(record, 'log_code') else None
+        log_code = record.log_code if hasattr(record, 'log_code') else None # type: ignore
         header = self._make_header(timestamp, channel, level, log_code)
         # Pretty format the message
         indent = self._INDENT*self._indent.indent
@@ -296,7 +299,7 @@ def format(self, record):
 g_alog_name_to_level = {name: level for level, name in g_alog_level_to_name.items()}
 
 # Global map of default formatters
-g_alog_formatters = {
+g_alog_formatters:dict[str, type[AlogFormatterBase]] = {
     "json": AlogJsonFormatter,
     "pretty": AlogPrettyFormatter,
 }
@@ -311,7 +314,7 @@ def format(self, record):
 
 # The current set of channel names that are managed via filters. This is
 # necessary to enable reconfiguring dynamically
-g_filtered_channels = []
+g_filtered_channels:list[str] = []
 
 class _MultiEqualString:
     """This 'str' class is used to allow the __eq__ operator to match multiple
@@ -319,7 +322,7 @@ class _MultiEqualString:
     this file matches True for == when checking if a stack frame matches an
     "internal" name.
     """
-    def __init__(self, *strings):
+    def __init__(self, *strings:str):
         self._strings = strings
 
     def __eq__(self, other):
@@ -327,10 +330,10 @@ def __eq__(self, other):
 
 # If this is python 3.8+, the _log function has a `stacklevel` argument that
 # can be given to indicate the need to pop additional levels off the stack.
-# This is the _right_ way to de-ailas the wrapper function, but it doesn't
+# This is the _right_ way to de-alias the wrapper function, but it doesn't
 # work on python 3.6 and 3.7.
-g_log_extra_kwargs = {}
-if sys.version_info >= (3, 8, 0, "", 0):
+g_log_extra_kwargs:dict[str, Any] = {}
+if sys.version_info >= (3, 8, 0, "", 0): # type: ignore
     # Pop 2 additional levels off the stack:
     #   - _log_with_code_method_override
     #   - inline lambda
@@ -344,10 +347,10 @@ def __eq__(self, other):
 else:
     logging._srcfile = _MultiEqualString(logging._srcfile, __file__)
 
-def is_log_code(arg):
+def is_log_code(arg:str):
     return arg.startswith('<') and arg.endswith('>')
 
-def _get_level_value(level_name):
+def _get_level_value(level_name: _Level):
     if isinstance(level_name, int):
         return level_name
     val = g_alog_name_to_level.get(level_name, None)
@@ -358,7 +361,7 @@ def _get_level_value(level_name):
     except ValueError:
         logging.warning("Invalid log level: %s", level_name)
 
-def _log_with_code_method_override(self, value, arg_one, *args, **kwargs):
+def _log_with_code_method_override(self: logging.Logger, value:int, arg_one, *args, **kwargs):
     """This helper is used as an override to the native logging.Logger instance
     methods for each level. As such, it's first argument, self, is the logger
     instance (or the global root logger singleton) on which to call the method.
@@ -394,7 +397,7 @@ def _log_with_code_method_override(self, value, arg_one, *args, **kwargs):
     else:
         self.log(value, arg_one, *args, **g_log_extra_kwargs, **kwargs)
 
-def _add_level_fn(name, value):
+def _add_level_fn(name:str, value:int):
     logging.addLevelName(value, name.upper())
 
     log_using_self_func = lambda self, arg_one, *args, **kwargs: \
@@ -403,11 +406,11 @@ def _add_level_fn(name, value):
     setattr(logging.Logger, name, log_using_self_func)
 
 def _add_is_enabled():
-    def is_enabled_func(self, level):
+    def is_enabled_func(self, level: _Level):
         return self.isEnabledFor(_get_level_value(level))
     setattr(logging.Logger, 'isEnabled', is_enabled_func)
 
-def _setup_formatter(formatter):
+def _setup_formatter(formatter:Union[str, AlogFormatterBase]):
     # If the formatter is a string, pull it from the defaults
     global g_alog_formatter
     if isinstance(formatter, str):
@@ -428,7 +431,8 @@ def _setup_formatter(formatter):
     else:
         raise ValueError("Invalid formatter type: %s" % type(formatter))
 
-def _parse_filters(filters):
+def _parse_filters(filters:Union[str, dict[str, _Level]]) -> dict[str, _Level]:
+    """Parse and remove filters with invalid log levels."""
     # Check to see if we've got a dictionary. If we do, keep the valid filter entries
     if isinstance(filters, dict):
         return _parse_dict_of_filters(filters)
@@ -441,15 +445,15 @@ def _parse_filters(filters):
         logging.warning("Invalid filter type [%s] was ignored!", type(filters).__name__)
         return {}
 
-def _parse_dict_of_filters(filters):
+def _parse_dict_of_filters(filters:dict[str, _Level]):
     for entry, level_name in filters.items():
         if _get_level_value(level_name) is None:
             logging.warning("Invalid filter entry [%s]", entry)
             del filters[entry]
     return filters
 
-def _parse_str_of_filters(filters):
-    chan_map = {}
+def _parse_str_of_filters(filters:str):
+    chan_map:dict[str, _Level] = {}
     for entry in filters.split(','):
         if len(entry):
             parts = entry.split(':')
@@ -479,11 +483,11 @@ def _parse_str_of_filters(filters):
 ## Core ########################################################################
 
 def configure(
-    default_level,
-    filters="",
-    formatter='pretty',
+    default_level:str,
+    filters:Union[str, dict[str, _Level]]="",
+    formatter:Union[str, AlogFormatterBase]='pretty',
     thread_id=False,
-    handler_generator=None,
+    handler_generator:Optional[Callable[[], logging.Handler]]=None,
 ):
     """Top-level configuration function for the alog module. This function
     configures the logging package to use the given default level and
@@ -554,7 +558,7 @@ def configure(
     else:
         logging.warning("Invalid default_level [%s]", default_level)
 
-    # Parse the filters
+    # Parse the filters and remove filters with invalid log levels
     parsed_filters = _parse_filters(filters)
 
     # There seems to be no way to reattach a logger to the root after being
@@ -566,9 +570,10 @@ def configure(
             parsed_filters[chan] = default_level
 
     # Add level filters by name
-    # NOTE: All levels assumed valid after call to _parse_filters
     for chan, level_name in parsed_filters.items():
         level = _get_level_value(level_name)
+        assert level is not None # All levels assumed valid after call to _parse_filters
+
         handler = handler_generator()
         handler.setFormatter(g_alog_formatter)
         handler.setLevel(level)
@@ -582,7 +587,7 @@ def configure(
     # Store the names of all channels currently managed by filters
     g_filtered_channels = list(parsed_filters.keys())
 
-def use_channel(channel):
+def use_channel(channel:Optional[str]):
     """Interface wrapper for python alog implementation to keep consistency with
     other languages.
     """
@@ -680,11 +685,11 @@ def __exit__(self, exception_type, exception_value, traceback):
 # pylint: disable=too-few-public-methods
 class FunctionLog(ScopedLog):
     """Function log behaves like a ScopedLog but adds the function name to the
-    begin and end messages.  This is intended to be used for loging when a
+    begin and end messages.  This is intended to be used for logging when a
     function starts and ends.
 
     Notes:
-        Using the @alog.logged_function decorator is the prefered (pythonic) method
+        Using the @alog.logged_function decorator is the preferred (pythonic) method
         for logging functions, consider using that instead.
 
     Examples:
@@ -704,7 +709,7 @@ class FnLog(FunctionLog):
 
 def logged_function(log_fn, format_str="", *fmt_args):
     """Function log decorator is a scoped log that adds the function name to the
-    begin and end messages.  This is intended to be used for loging when a
+    begin and end messages.  This is intended to be used for logging when a
     function starts and ends.
 
     Examples:
@@ -759,7 +764,7 @@ class ScopedTimer(_TimedLogBase):
     at destruction.
 
     Notes:
-        Using the @alog.timed_function decorator is the prefered (pythonic)
+        Using the @alog.timed_function decorator is the preferred (pythonic)
         method for timing entire functions, consider using that instead.
 
     Examples:
@@ -801,7 +806,7 @@ def __exit__(self, exception_type, exception_value, traceback):
 
 def timed_function(log_fn, format_str="", *fmt_args):
     """Timed function decorator is a scoped timer that adds the function name to
-    the end messages.  This is intended to be used for loging the time required
+    the end messages.  This is intended to be used for logging the time required
     for a function to complete.
 
     Examples:

From 1b036a96b0d6bbc12050d2362a4204ca61c96443 Mon Sep 17 00:00:00 2001
From: Mynhardt Burger <mynhardt@gmail.com>
Date: Sun, 28 Jan 2024 16:39:31 -0500
Subject: [PATCH 2/7] Spelling

Signed-off-by: Mynhardt Burger <mynhardt@gmail.com>
---
 src/python/alog/alog.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/python/alog/alog.py b/src/python/alog/alog.py
index 282e1ec..f24c692 100644
--- a/src/python/alog/alog.py
+++ b/src/python/alog/alog.py
@@ -71,7 +71,7 @@ def __init__(self):
 
     def formatTime(self, record:logging.LogRecord, datefmt:Optional[str]=None):
         """A wrapper for the parent formatTime that returns UTC timezone
-        time stamp inISO format.
+        time stamp in ISO format.
 
         Args:
             record (LogRecord):  Log record to pull the created date from.

From 39cd7932b72875e68315f46a521613ff6c7fdbd7 Mon Sep 17 00:00:00 2001
From: Mynhardt Burger <mynhardt@gmail.com>
Date: Mon, 29 Jan 2024 13:16:27 -0500
Subject: [PATCH 3/7] Black formatting

Signed-off-by: Mynhardt Burger <mynhardt@gmail.com>
---
 src/python/alog/alog.py | 322 ++++++++++++++++++++++++----------------
 1 file changed, 193 insertions(+), 129 deletions(-)

diff --git a/src/python/alog/alog.py b/src/python/alog/alog.py
index f24c692..b440e13 100644
--- a/src/python/alog/alog.py
+++ b/src/python/alog/alog.py
@@ -42,14 +42,15 @@
 
 g_thread_id_enabled = False
 
+
 class AlogFormatterBase(logging.Formatter):
-    """Base class with common functionality for alog formatters.
-    """
+    """Base class with common functionality for alog formatters."""
 
     class ThreadLocalIndent(threading.local):
-        '''Private subclass of threading.local which initializes to 0 on
+        """Private subclass of threading.local which initializes to 0 on
         construction
-        '''
+        """
+
         def __init__(self):
             self.indent = 0
 
@@ -69,7 +70,7 @@ def __init__(self):
         # Initialize the underlying logger with this formatter
         logging.Formatter.__init__(self)
 
-    def formatTime(self, record:logging.LogRecord, datefmt:Optional[str]=None):
+    def formatTime(self, record: logging.LogRecord, datefmt: Optional[str] = None):
         """A wrapper for the parent formatTime that returns UTC timezone
         time stamp in ISO format.
 
@@ -83,40 +84,49 @@ def formatTime(self, record:logging.LogRecord, datefmt:Optional[str]=None):
         return datetime.utcfromtimestamp(record.created).isoformat()
 
     def indent(self):
-        """Add a level of indentation for this thread.
-        """
+        """Add a level of indentation for this thread."""
         self._indent.indent += 1
 
     def deindent(self):
-        """Remove a level of indentation for this thread.
-        """
+        """Remove a level of indentation for this thread."""
         if self._indent.indent > 0:
             self._indent.indent -= 1
 
+
 class AlogJsonFormatter(AlogFormatterBase):
-    """Log formatter which prints messages a single-line json.
-    """
-    _FIELDS_TO_PRINT = ['name', 'levelname', 'asctime', 'message',
-                        'exc_text', 'region-id', 'org-id', 'tran-id',
-                        'watson-txn-id', 'channel', 'duration']
+    """Log formatter which prints messages a single-line json."""
+
+    _FIELDS_TO_PRINT = [
+        "name",
+        "levelname",
+        "asctime",
+        "message",
+        "exc_text",
+        "region-id",
+        "org-id",
+        "tran-id",
+        "watson-txn-id",
+        "channel",
+        "duration",
+    ]
 
     def __init__(self):
         AlogFormatterBase.__init__(self)
 
     @staticmethod
-    def _map_to_common_key_name(log_record_keyname:str):
-        if log_record_keyname == 'levelname':
-            return 'level'
-        elif log_record_keyname == 'asctime':
-            return 'timestamp'
-        elif log_record_keyname == 'exc_text':
-            return 'exception'
-        elif log_record_keyname == 'name':
-            return 'channel'
+    def _map_to_common_key_name(log_record_keyname: str):
+        if log_record_keyname == "levelname":
+            return "level"
+        elif log_record_keyname == "asctime":
+            return "timestamp"
+        elif log_record_keyname == "exc_text":
+            return "exception"
+        elif log_record_keyname == "name":
+            return "channel"
         else:
             return log_record_keyname
 
-    def _extract_fields_from_record_as_dict(self, record:logging.LogRecord):
+    def _extract_fields_from_record_as_dict(self, record: logging.LogRecord):
         """Extracts the fields we want out of log record and puts them into an
         dict for easy jsonification.
 
@@ -127,7 +137,7 @@ def _extract_fields_from_record_as_dict(self, record:logging.LogRecord):
             The relevant fields pulled out from the log record object and
             initialized into a dictionary.
         """
-        out:dict[Any, Any] = {}
+        out: dict[Any, Any] = {}
         for field_name in self._FIELDS_TO_PRINT:
             if hasattr(record, field_name):
                 record_field = getattr(record, field_name)
@@ -139,19 +149,19 @@ def _extract_fields_from_record_as_dict(self, record:logging.LogRecord):
         out["level"] = out["level"].lower()
         return out
 
-    def format(self, record:logging.LogRecord):
+    def format(self, record: logging.LogRecord):
         """Formats the log record as a JSON formatted string
-       (also removes new line characters so everything prints on a single line)
+        (also removes new line characters so everything prints on a single line)
 
-        Args:
-            record (logging.LogRecord):  The record to extract from.
+         Args:
+             record (logging.LogRecord):  The record to extract from.
 
-        Returns:
-            The jsonified string representation of the record.
+         Returns:
+             The jsonified string representation of the record.
         """
         # Maintain the message as a dict if passed in as one
         if isinstance(record.msg, dict):
-            record.message = record.msg # type: ignore
+            record.message = record.msg  # type: ignore
         else:
             record.message = record.getMessage()
 
@@ -165,26 +175,27 @@ def format(self, record:logging.LogRecord):
         log_record = self._extract_fields_from_record_as_dict(record)
 
         # Add indent to all log records
-        log_record['num_indent'] = self._indent.indent
+        log_record["num_indent"] = self._indent.indent
 
         # If enabled, add thread id
         if g_thread_id_enabled:
-            log_record['thread_id'] = threading.get_ident()
+            log_record["thread_id"] = threading.get_ident()
 
         # Interpolate message and args if present
-        record_args = log_record.pop('args', None)
+        record_args = log_record.pop("args", None)
         if record_args:
-            message = log_record.get('message')
+            message = log_record.get("message")
             if message:
-                log_record['message'] = message % record_args
+                log_record["message"] = message % record_args
             else:
-                log_record['message'] = str(record_args)
+                log_record["message"] = str(record_args)
 
         return json.dumps(log_record, sort_keys=True)
 
+
 class AlogPrettyFormatter(AlogFormatterBase):
-    """Log formatter that pretty-prints lines for easy visibility.
-    """
+    """Log formatter that pretty-prints lines for easy visibility."""
+
     _INDENT = "  "
     _LEVEL_MAP = {
         "critical": "FATL",
@@ -204,16 +215,17 @@ def __init__(self, channel_len=5):
         AlogFormatterBase.__init__(self)
         self.channel_len = channel_len
 
-    def _make_header(self, timestamp:str, channel:str, level:str, log_code:Optional[str]):
-        """Create the header for a log line with proper padding.
-        """
+    def _make_header(
+        self, timestamp: str, channel: str, level: str, log_code: Optional[str]
+    ):
+        """Create the header for a log line with proper padding."""
         # Get the padded or truncated channel
         chan = channel
         if len(channel) > self.channel_len:
-            chan = channel[:self.channel_len]
+            chan = channel[: self.channel_len]
 
         elif len(channel) < self.channel_len:
-            chan = channel + ' ' * (self.channel_len - len(channel))
+            chan = channel + " " * (self.channel_len - len(channel))
 
         # Get the mapped level
         lvl = self._LEVEL_MAP.get(level.lower(), "UNKN")
@@ -230,7 +242,7 @@ def _make_header(self, timestamp:str, channel:str, level:str, log_code:Optional[
 
         return header
 
-    def format(self, record:logging.LogRecord):
+    def format(self, record: logging.LogRecord):
         """Formats the log record as pretty-printed lines of the format:
 
         timestamp [CHANL:LEVL] message
@@ -238,60 +250,73 @@ def format(self, record:logging.LogRecord):
         # Extract special values from the message if it's a dict
         metadata = None
         if isinstance(record.msg, dict):
-            if 'message' in record.msg:
-                message = record.msg.pop('message')
-                args = record.msg.pop('args', None)
+            if "message" in record.msg:
+                message = record.msg.pop("message")
+                args = record.msg.pop("args", None)
                 if args:
                     message = message % args
                 record.message = message
-            if 'log_code' in record.msg:
-                record.log_code = record.msg.pop('log_code')
+            if "log_code" in record.msg:
+                record.log_code = record.msg.pop("log_code")
             metadata = record.msg
         else:
             record.message = record.getMessage()
 
         # Add metadata if present
-        if not hasattr(record, 'message'):
-            record.message = ''
+        if not hasattr(record, "message"):
+            record.message = ""
         if metadata is not None and len(metadata) > 0:
             if len(record.message) > 0:
-                record.message += ' '
+                record.message += " "
             record.message += json.dumps(metadata)
 
         level = record.levelname
         channel = record.name
         timestamp = self.formatTime(record, self.datefmt)
-        log_code = record.log_code if hasattr(record, 'log_code') else None # type: ignore
+        log_code = record.log_code if hasattr(record, "log_code") else None  # type: ignore
         header = self._make_header(timestamp, channel, level, log_code)
         # Pretty format the message
-        indent = self._INDENT*self._indent.indent
+        indent = self._INDENT * self._indent.indent
         if isinstance(record.message, str):
-            formatted = ['%s %s%s' % (header, indent, line) for line in record.message.split('\n')]
+            formatted = [
+                "%s %s%s" % (header, indent, line)
+                for line in record.message.split("\n")
+            ]
         else:
-            formatted = ['%s %s%s' % (header, indent, str(record.message))]
+            formatted = ["%s %s%s" % (header, indent, str(record.message))]
 
         # Add stack trace if present
         if record.exc_info:
-            formatted.extend(['%s %s%s' % (header, indent, line) for line in self.formatException(record.exc_info).split('\n')])
-
-        formatted = '\n'.join(formatted)
+            formatted.extend(
+                [
+                    "%s %s%s" % (header, indent, line)
+                    for line in self.formatException(record.exc_info).split("\n")
+                ]
+            )
+
+        formatted = "\n".join(formatted)
         return formatted
 
+
 ## Constants ###################################################################
 
 # Global maps from name <-> level, pull from logging packages for consistency
 # pylint: disable=protected-access
-g_alog_level_to_name = {level: name.lower() for level, name in logging._levelToName.items()}
+g_alog_level_to_name = {
+    level: name.lower() for level, name in logging._levelToName.items()
+}
 
 # Extra custom log levels
-g_alog_level_to_name.update({
-    60: "off",
-    15: "trace",
-    9: "debug1",
-    8: "debug2",
-    7: "debug3",
-    6: "debug4",
-})
+g_alog_level_to_name.update(
+    {
+        60: "off",
+        15: "trace",
+        9: "debug1",
+        8: "debug2",
+        7: "debug3",
+        6: "debug4",
+    }
+)
 
 # Special "level" used to disable all logging
 g_disable_level = "disable"
@@ -299,7 +324,7 @@ def format(self, record:logging.LogRecord):
 g_alog_name_to_level = {name: level for level, name in g_alog_level_to_name.items()}
 
 # Global map of default formatters
-g_alog_formatters:dict[str, type[AlogFormatterBase]] = {
+g_alog_formatters: dict[str, type[AlogFormatterBase]] = {
     "json": AlogJsonFormatter,
     "pretty": AlogPrettyFormatter,
 }
@@ -314,7 +339,8 @@ def format(self, record:logging.LogRecord):
 
 # The current set of channel names that are managed via filters. This is
 # necessary to enable reconfiguring dynamically
-g_filtered_channels:list[str] = []
+g_filtered_channels: list[str] = []
+
 
 class _MultiEqualString:
     """This 'str' class is used to allow the __eq__ operator to match multiple
@@ -322,18 +348,20 @@ class _MultiEqualString:
     this file matches True for == when checking if a stack frame matches an
     "internal" name.
     """
-    def __init__(self, *strings:str):
+
+    def __init__(self, *strings: str):
         self._strings = strings
 
     def __eq__(self, other):
         return other in self._strings
 
+
 # If this is python 3.8+, the _log function has a `stacklevel` argument that
 # can be given to indicate the need to pop additional levels off the stack.
 # This is the _right_ way to de-alias the wrapper function, but it doesn't
 # work on python 3.6 and 3.7.
-g_log_extra_kwargs:dict[str, Any] = {}
-if sys.version_info >= (3, 8, 0, "", 0): # type: ignore
+g_log_extra_kwargs: dict[str, Any] = {}
+if sys.version_info >= (3, 8, 0, "", 0):  # type: ignore
     # Pop 2 additional levels off the stack:
     #   - _log_with_code_method_override
     #   - inline lambda
@@ -347,8 +375,10 @@ def __eq__(self, other):
 else:
     logging._srcfile = _MultiEqualString(logging._srcfile, __file__)
 
-def is_log_code(arg:str):
-    return arg.startswith('<') and arg.endswith('>')
+
+def is_log_code(arg: str):
+    return arg.startswith("<") and arg.endswith(">")
+
 
 def _get_level_value(level_name: _Level):
     if isinstance(level_name, int):
@@ -361,7 +391,10 @@ def _get_level_value(level_name: _Level):
     except ValueError:
         logging.warning("Invalid log level: %s", level_name)
 
-def _log_with_code_method_override(self: logging.Logger, value:int, arg_one, *args, **kwargs):
+
+def _log_with_code_method_override(
+    self: logging.Logger, value: int, arg_one, *args, **kwargs
+):
     """This helper is used as an override to the native logging.Logger instance
     methods for each level. As such, it's first argument, self, is the logger
     instance (or the global root logger singleton) on which to call the method.
@@ -397,20 +430,27 @@ def _log_with_code_method_override(self: logging.Logger, value:int, arg_one, *ar
     else:
         self.log(value, arg_one, *args, **g_log_extra_kwargs, **kwargs)
 
-def _add_level_fn(name:str, value:int):
+
+def _add_level_fn(name: str, value: int):
     logging.addLevelName(value, name.upper())
 
-    log_using_self_func = lambda self, arg_one, *args, **kwargs: \
-        _log_with_code_method_override(self, value, arg_one, *args, **kwargs)
-    setattr(log_using_self_func, '_level_value', value)
+    log_using_self_func = (
+        lambda self, arg_one, *args, **kwargs: _log_with_code_method_override(
+            self, value, arg_one, *args, **kwargs
+        )
+    )
+    setattr(log_using_self_func, "_level_value", value)
     setattr(logging.Logger, name, log_using_self_func)
 
+
 def _add_is_enabled():
     def is_enabled_func(self, level: _Level):
         return self.isEnabledFor(_get_level_value(level))
-    setattr(logging.Logger, 'isEnabled', is_enabled_func)
 
-def _setup_formatter(formatter:Union[str, AlogFormatterBase]):
+    setattr(logging.Logger, "isEnabled", is_enabled_func)
+
+
+def _setup_formatter(formatter: Union[str, AlogFormatterBase]):
     # If the formatter is a string, pull it from the defaults
     global g_alog_formatter
     if isinstance(formatter, str):
@@ -431,7 +471,8 @@ def _setup_formatter(formatter:Union[str, AlogFormatterBase]):
     else:
         raise ValueError("Invalid formatter type: %s" % type(formatter))
 
-def _parse_filters(filters:Union[str, dict[str, _Level]]) -> dict[str, _Level]:
+
+def _parse_filters(filters: Union[str, dict[str, _Level]]) -> dict[str, _Level]:
     """Parse and remove filters with invalid log levels."""
     # Check to see if we've got a dictionary. If we do, keep the valid filter entries
     if isinstance(filters, dict):
@@ -445,31 +486,36 @@ def _parse_filters(filters:Union[str, dict[str, _Level]]) -> dict[str, _Level]:
         logging.warning("Invalid filter type [%s] was ignored!", type(filters).__name__)
         return {}
 
-def _parse_dict_of_filters(filters:dict[str, _Level]):
+
+def _parse_dict_of_filters(filters: dict[str, _Level]):
     for entry, level_name in filters.items():
         if _get_level_value(level_name) is None:
             logging.warning("Invalid filter entry [%s]", entry)
             del filters[entry]
     return filters
 
-def _parse_str_of_filters(filters:str):
-    chan_map:dict[str, _Level] = {}
-    for entry in filters.split(','):
+
+def _parse_str_of_filters(filters: str):
+    chan_map: dict[str, _Level] = {}
+    for entry in filters.split(","):
         if len(entry):
-            parts = entry.split(':')
+            parts = entry.split(":")
             if len(parts) != 2:
                 logging.warning("Invalid filter entry [%s]", entry)
             else:
                 chan, level_name = parts
                 level = _get_level_value(level_name)
                 if level is None:
-                    logging.warning("Invalid level [%s] for channel [%s]", level_name, chan)
+                    logging.warning(
+                        "Invalid level [%s] for channel [%s]", level_name, chan
+                    )
                 else:
                     chan_map[chan] = level_name
         else:
             logging.warning("Invalid filter entry [%s]", entry)
     return chan_map
 
+
 ## Import-time Setup ###########################################################
 
 # Add custom low levels
@@ -482,12 +528,13 @@ def _parse_str_of_filters(filters:str):
 
 ## Core ########################################################################
 
+
 def configure(
-    default_level:str,
-    filters:Union[str, dict[str, _Level]]="",
-    formatter:Union[str, AlogFormatterBase]='pretty',
+    default_level: str,
+    filters: Union[str, dict[str, _Level]] = "",
+    formatter: Union[str, AlogFormatterBase] = "pretty",
     thread_id=False,
-    handler_generator:Optional[Callable[[], logging.Handler]]=None,
+    handler_generator: Optional[Callable[[], logging.Handler]] = None,
 ):
     """Top-level configuration function for the alog module. This function
     configures the logging package to use the given default level and
@@ -572,7 +619,9 @@ def configure(
     # Add level filters by name
     for chan, level_name in parsed_filters.items():
         level = _get_level_value(level_name)
-        assert level is not None # All levels assumed valid after call to _parse_filters
+        assert (
+            level is not None
+        )  # All levels assumed valid after call to _parse_filters
 
         handler = handler_generator()
         handler.setFormatter(g_alog_formatter)
@@ -587,14 +636,17 @@ def configure(
     # Store the names of all channels currently managed by filters
     g_filtered_channels = list(parsed_filters.keys())
 
-def use_channel(channel:Optional[str]):
+
+def use_channel(channel: Optional[str]):
     """Interface wrapper for python alog implementation to keep consistency with
     other languages.
     """
     return logging.getLogger(channel)
 
+
 ## Scoped Loggers ##############################################################
 
+
 # The whole point of this class is act on scope changes
 # pylint: disable=too-few-public-methods
 class _ScopedLogBase:
@@ -602,9 +654,9 @@ class _ScopedLogBase:
     and stopping the logger and expects the child class to call them when
     appropriate.
     """
+
     def __init__(self, log_fn, format_str="", *args):
-        """Construct a new scoped logger.
-        """
+        """Construct a new scoped logger."""
         self.log_fn = log_fn
         self.format_str = format_str
         self.args = args
@@ -617,13 +669,13 @@ def __init__(self, log_fn, format_str="", *args):
         # 1. Get the parent channel (logging.Logger) instance from __self__
         # 2. Get the numeric level value of the bound function from _level_value
         # 3. Check if that level value is enabled for that logger
-        assert hasattr(self.log_fn, '_level_value'), \
-            'Cannot use non-logging function for scoped log'
+        assert hasattr(
+            self.log_fn, "_level_value"
+        ), "Cannot use non-logging function for scoped log"
         self.enabled = self.log_fn.__self__.isEnabledFor(self.log_fn._level_value)
 
     def _start_scoped_log(self):
-        """Log the start message for a scoped logger and increment the indentor.
-        """
+        """Log the start message for a scoped logger and increment the indentor."""
         if self.enabled:
             self.log_fn(scope_start_str + str(self.format_str), *self.args)
             global g_alog_formatter
@@ -631,14 +683,14 @@ def _start_scoped_log(self):
                 g_alog_formatter.indent()
 
     def _end_scoped_log(self):
-        """Log the end message for a scoped logger and decrement the indentor.
-        """
+        """Log the end message for a scoped logger and decrement the indentor."""
         if self.enabled:
             global g_alog_formatter
             if g_alog_formatter:
                 g_alog_formatter.deindent()
             self.log_fn(scope_end_str + str(self.format_str), *self.args)
 
+
 # pylint: disable=too-few-public-methods
 class ScopedLog(_ScopedLogBase):
     """Scoped log prints a begin message when constructed and an end message on
@@ -649,17 +701,17 @@ class ScopedLog(_ScopedLogBase):
         >>>     # will log begin message here and end message after returning
         >>>     _ = alog.ScopedLog(log_channel.debug)
     """
+
     def __init__(self, log_fn, format_str="", *args):
-        """Construct a new scoped logger and print the begin message.
-        """
+        """Construct a new scoped logger and print the begin message."""
         super().__init__(log_fn, format_str, *args)
         self._start_scoped_log()
 
     def __del__(self):
-        """Print the end message when this logger is deleted.
-        """
+        """Print the end message when this logger is deleted."""
         self._end_scoped_log()
 
+
 # pylint: disable=too-few-public-methods
 class ContextLog(_ScopedLogBase):
     """Context log prints a begin message when a context manager is entered and
@@ -671,17 +723,17 @@ class ContextLog(_ScopedLogBase):
         >>>   print('hello world')
         >>>   # logs the end message when the context manager exits
     """
+
     def __enter__(self):
-        """Log the begin message when the context manager starts.
-        """
+        """Log the begin message when the context manager starts."""
         self._start_scoped_log()
         return self
 
     def __exit__(self, exception_type, exception_value, traceback):
-        """Log the end message when the context manager exits.
-        """
+        """Log the end message when the context manager exits."""
         self._end_scoped_log()
 
+
 # pylint: disable=too-few-public-methods
 class FunctionLog(ScopedLog):
     """Function log behaves like a ScopedLog but adds the function name to the
@@ -698,15 +750,20 @@ class FunctionLog(ScopedLog):
         >>>     # function returns messages will include the name test_function
         >>>     _ = alog.FunctionLog(log_channel.debug)
     """
+
     def __init__(self, log_fn, format_str="", *args):
-        fn_name = traceback.format_stack()[-2].strip().split(',')[2].split(' ')[2].strip()
+        fn_name = (
+            traceback.format_stack()[-2].strip().split(",")[2].split(" ")[2].strip()
+        )
         format_str = "%s(" + format_str + ")"
         super().__init__(log_fn, format_str, fn_name, *args)
 
+
 # older name for FunctionLog, provided for compatibility
 class FnLog(FunctionLog):
     pass
 
+
 def logged_function(log_fn, format_str="", *fmt_args):
     """Function log decorator is a scoped log that adds the function name to the
     begin and end messages.  This is intended to be used for logging when a
@@ -719,45 +776,49 @@ def logged_function(log_fn, format_str="", *fmt_args):
         >>>     # the end message after the function exits
         >>>     print('hello world!')
     """
+
     # decorator function returned after arguments are passed
     def decorator(func):
         # wrapper function returned by decorator
-        @functools.wraps(func) # ensures that docstrings are maintained
+        @functools.wraps(func)  # ensures that docstrings are maintained
         def wrapper(*args, **kwargs):
             fmt_str = "%s(" + format_str + ")"
             with ContextLog(log_fn, fmt_str, func.__name__, *fmt_args):
                 return func(*args, **kwargs)
+
         return wrapper
+
     return decorator
 
+
 ## Timers ######################################################################
 
+
 class _TimedLogBase:
     """Base class for timed loggers.  This class provides methods for starting
     and stopping the logger and expects the child class to call them when
     appropriate.
     """
+
     def __init__(self, log_fn, format_str="", *args):
-        """Construct a new timed logger.
-        """
+        """Construct a new timed logger."""
         self.log_fn = log_fn
         self.format_str = format_str
         self.args = args
         self.start_time = 0
 
     def _start_timed_log(self):
-        """Get the start time for this timed logger.
-        """
+        """Get the start time for this timed logger."""
         self.start_time = time.time()
 
     def _end_timed_log(self):
-        """Gets the end time and prints the end message for this timed logger.
-        """
+        """Gets the end time and prints the end message for this timed logger."""
         duration = timedelta(seconds=time.time() - self.start_time)
         fmt = self.format_str + "%s"
         args = list(self.args) + [str(duration)]
         self.log_fn(fmt, *args, extra={"duration": duration.total_seconds()})
 
+
 # pylint: disable=too-few-public-methods
 class ScopedTimer(_TimedLogBase):
     """Scoped timer that starts a timer at construction and logs the time delta
@@ -772,17 +833,17 @@ class ScopedTimer(_TimedLogBase):
         >>>     # will log the time delta when the function exits
         >>>     _ = alog.ScopedTimer(log_channel.debug)
     """
+
     def __init__(self, log_fn, format_str="", *args):
-        """Construct a new scoped timer and get the start time.
-        """
+        """Construct a new scoped timer and get the start time."""
         super().__init__(log_fn, format_str, *args)
         self._start_timed_log()
 
     def __del__(self):
-        """Log the end message, including time delta, when this timer is deleted.
-        """
+        """Log the end message, including time delta, when this timer is deleted."""
         self._end_timed_log()
 
+
 class ContextTimer(_TimedLogBase):
     """Context timer that starts a timer when a context is entered and logs the
     time delta when the context exits.
@@ -793,17 +854,17 @@ class ContextTimer(_TimedLogBase):
         >>>   print('hello world')
         >>>   # logs the time delta when the context manager exits
     """
+
     def __enter__(self):
-        """Start the timer when a context is entered.
-        """
+        """Start the timer when a context is entered."""
         self._start_timed_log()
         return self
 
     def __exit__(self, exception_type, exception_value, traceback):
-        """Log the end message, including time delta, when the context exits.
-        """
+        """Log the end message, including time delta, when the context exits."""
         self._end_timed_log()
 
+
 def timed_function(log_fn, format_str="", *fmt_args):
     """Timed function decorator is a scoped timer that adds the function name to
     the end messages.  This is intended to be used for logging the time required
@@ -816,6 +877,7 @@ def timed_function(log_fn, format_str="", *fmt_args):
         >>>     # the time spent after it returns
         >>>     print('hello world!')
     """
+
     # decorator function returned after arguments are passed
     def decorator(func):
         # wrapper function returned by decorator
@@ -823,5 +885,7 @@ def decorator(func):
         def wrapper(*args, **kwargs):
             with ContextTimer(log_fn, format_str, *fmt_args):
                 return func(*args, **kwargs)
+
         return wrapper
+
     return decorator

From aca64a34ede0b3703f19ced08255baff2a299b96 Mon Sep 17 00:00:00 2001
From: Mynhardt Burger <mynhardt@gmail.com>
Date: Mon, 29 Jan 2024 18:52:51 -0500
Subject: [PATCH 4/7] Add return types, use typing.List/Dict, import sorting

Signed-off-by: Mynhardt Burger <mynhardt@gmail.com>
---
 src/python/alog/alog.py | 108 +++++++++++++++++++++-------------------
 1 file changed, 56 insertions(+), 52 deletions(-)

diff --git a/src/python/alog/alog.py b/src/python/alog/alog.py
index b440e13..83f6282 100644
--- a/src/python/alog/alog.py
+++ b/src/python/alog/alog.py
@@ -34,7 +34,7 @@
 import time
 import traceback
 from datetime import datetime, timedelta
-from typing import Any, Callable, Union, Optional
+from typing import Any, Callable, Dict, List, Optional, Union
 
 _Level = Union[int, str]
 
@@ -51,16 +51,16 @@ class ThreadLocalIndent(threading.local):
         construction
         """
 
-        def __init__(self):
+        def __init__(self) -> None:
             self.indent = 0
 
-        def __getstate__(self):
+        def __getstate__(self) -> None:
             return None
 
-        def __setstate__(self, d):
+        def __setstate__(self, d) -> None:
             pass
 
-    def __init__(self):
+    def __init__(self) -> None:
         # Hold a thread-local map for storing indentation so that the counts are
         # kept independently for each thread. Note that threading.local values
         # are cleaned up when their local thread dies, so this is safe to use
@@ -70,7 +70,9 @@ def __init__(self):
         # Initialize the underlying logger with this formatter
         logging.Formatter.__init__(self)
 
-    def formatTime(self, record: logging.LogRecord, datefmt: Optional[str] = None):
+    def formatTime(
+        self, record: logging.LogRecord, datefmt: Optional[str] = None
+    ) -> str:
         """A wrapper for the parent formatTime that returns UTC timezone
         time stamp in ISO format.
 
@@ -83,11 +85,11 @@ def formatTime(self, record: logging.LogRecord, datefmt: Optional[str] = None):
         """
         return datetime.utcfromtimestamp(record.created).isoformat()
 
-    def indent(self):
+    def indent(self) -> None:
         """Add a level of indentation for this thread."""
         self._indent.indent += 1
 
-    def deindent(self):
+    def deindent(self) -> None:
         """Remove a level of indentation for this thread."""
         if self._indent.indent > 0:
             self._indent.indent -= 1
@@ -96,7 +98,7 @@ def deindent(self):
 class AlogJsonFormatter(AlogFormatterBase):
     """Log formatter which prints messages a single-line json."""
 
-    _FIELDS_TO_PRINT = [
+    _FIELDS_TO_PRINT: List[str] = [
         "name",
         "levelname",
         "asctime",
@@ -110,11 +112,11 @@ class AlogJsonFormatter(AlogFormatterBase):
         "duration",
     ]
 
-    def __init__(self):
+    def __init__(self) -> None:
         AlogFormatterBase.__init__(self)
 
     @staticmethod
-    def _map_to_common_key_name(log_record_keyname: str):
+    def _map_to_common_key_name(log_record_keyname: str) -> str:
         if log_record_keyname == "levelname":
             return "level"
         elif log_record_keyname == "asctime":
@@ -137,7 +139,7 @@ def _extract_fields_from_record_as_dict(self, record: logging.LogRecord):
             The relevant fields pulled out from the log record object and
             initialized into a dictionary.
         """
-        out: dict[Any, Any] = {}
+        out: Dict[Any, Any] = {}
         for field_name in self._FIELDS_TO_PRINT:
             if hasattr(record, field_name):
                 record_field = getattr(record, field_name)
@@ -149,7 +151,7 @@ def _extract_fields_from_record_as_dict(self, record: logging.LogRecord):
         out["level"] = out["level"].lower()
         return out
 
-    def format(self, record: logging.LogRecord):
+    def format(self, record: logging.LogRecord) -> str:
         """Formats the log record as a JSON formatted string
         (also removes new line characters so everything prints on a single line)
 
@@ -197,7 +199,7 @@ class AlogPrettyFormatter(AlogFormatterBase):
     """Log formatter that pretty-prints lines for easy visibility."""
 
     _INDENT = "  "
-    _LEVEL_MAP = {
+    _LEVEL_MAP: Dict[str, str] = {
         "critical": "FATL",
         "fatal": "FATL",
         "error": "ERRR",
@@ -211,13 +213,13 @@ class AlogPrettyFormatter(AlogFormatterBase):
         "debug4": "DBG4",
     }
 
-    def __init__(self, channel_len=5):
+    def __init__(self, channel_len=5) -> None:
         AlogFormatterBase.__init__(self)
         self.channel_len = channel_len
 
     def _make_header(
         self, timestamp: str, channel: str, level: str, log_code: Optional[str]
-    ):
+    ) -> str:
         """Create the header for a log line with proper padding."""
         # Get the padded or truncated channel
         chan = channel
@@ -242,7 +244,7 @@ def _make_header(
 
         return header
 
-    def format(self, record: logging.LogRecord):
+    def format(self, record: logging.LogRecord) -> str:
         """Formats the log record as pretty-printed lines of the format:
 
         timestamp [CHANL:LEVL] message
@@ -302,7 +304,7 @@ def format(self, record: logging.LogRecord):
 
 # Global maps from name <-> level, pull from logging packages for consistency
 # pylint: disable=protected-access
-g_alog_level_to_name = {
+g_alog_level_to_name: Dict[int, str] = {
     level: name.lower() for level, name in logging._levelToName.items()
 }
 
@@ -321,10 +323,12 @@ def format(self, record: logging.LogRecord):
 # Special "level" used to disable all logging
 g_disable_level = "disable"
 
-g_alog_name_to_level = {name: level for level, name in g_alog_level_to_name.items()}
+g_alog_name_to_level: Dict[str, int] = {
+    name: level for level, name in g_alog_level_to_name.items()
+}
 
 # Global map of default formatters
-g_alog_formatters: dict[str, type[AlogFormatterBase]] = {
+g_alog_formatters: Dict[str, type[AlogFormatterBase]] = {
     "json": AlogJsonFormatter,
     "pretty": AlogPrettyFormatter,
 }
@@ -339,7 +343,7 @@ def format(self, record: logging.LogRecord):
 
 # The current set of channel names that are managed via filters. This is
 # necessary to enable reconfiguring dynamically
-g_filtered_channels: list[str] = []
+g_filtered_channels: List[str] = []
 
 
 class _MultiEqualString:
@@ -349,10 +353,10 @@ class _MultiEqualString:
     "internal" name.
     """
 
-    def __init__(self, *strings: str):
+    def __init__(self, *strings: str) -> None:
         self._strings = strings
 
-    def __eq__(self, other):
+    def __eq__(self, other) -> bool:
         return other in self._strings
 
 
@@ -360,7 +364,7 @@ def __eq__(self, other):
 # can be given to indicate the need to pop additional levels off the stack.
 # This is the _right_ way to de-alias the wrapper function, but it doesn't
 # work on python 3.6 and 3.7.
-g_log_extra_kwargs: dict[str, Any] = {}
+g_log_extra_kwargs: Dict[str, Any] = {}
 if sys.version_info >= (3, 8, 0, "", 0):  # type: ignore
     # Pop 2 additional levels off the stack:
     #   - _log_with_code_method_override
@@ -376,11 +380,11 @@ def __eq__(self, other):
     logging._srcfile = _MultiEqualString(logging._srcfile, __file__)
 
 
-def is_log_code(arg: str):
+def is_log_code(arg: str) -> bool:
     return arg.startswith("<") and arg.endswith(">")
 
 
-def _get_level_value(level_name: _Level):
+def _get_level_value(level_name: _Level) -> Optional[int]:
     if isinstance(level_name, int):
         return level_name
     val = g_alog_name_to_level.get(level_name, None)
@@ -394,7 +398,7 @@ def _get_level_value(level_name: _Level):
 
 def _log_with_code_method_override(
     self: logging.Logger, value: int, arg_one, *args, **kwargs
-):
+) -> None:
     """This helper is used as an override to the native logging.Logger instance
     methods for each level. As such, it's first argument, self, is the logger
     instance (or the global root logger singleton) on which to call the method.
@@ -431,7 +435,7 @@ def _log_with_code_method_override(
         self.log(value, arg_one, *args, **g_log_extra_kwargs, **kwargs)
 
 
-def _add_level_fn(name: str, value: int):
+def _add_level_fn(name: str, value: int) -> None:
     logging.addLevelName(value, name.upper())
 
     log_using_self_func = (
@@ -443,14 +447,14 @@ def _add_level_fn(name: str, value: int):
     setattr(logging.Logger, name, log_using_self_func)
 
 
-def _add_is_enabled():
+def _add_is_enabled() -> None:
     def is_enabled_func(self, level: _Level):
         return self.isEnabledFor(_get_level_value(level))
 
     setattr(logging.Logger, "isEnabled", is_enabled_func)
 
 
-def _setup_formatter(formatter: Union[str, AlogFormatterBase]):
+def _setup_formatter(formatter: Union[str, AlogFormatterBase]) -> None:
     # If the formatter is a string, pull it from the defaults
     global g_alog_formatter
     if isinstance(formatter, str):
@@ -472,7 +476,7 @@ def _setup_formatter(formatter: Union[str, AlogFormatterBase]):
         raise ValueError("Invalid formatter type: %s" % type(formatter))
 
 
-def _parse_filters(filters: Union[str, dict[str, _Level]]) -> dict[str, _Level]:
+def _parse_filters(filters: Union[str, Dict[str, _Level]]) -> Dict[str, _Level]:
     """Parse and remove filters with invalid log levels."""
     # Check to see if we've got a dictionary. If we do, keep the valid filter entries
     if isinstance(filters, dict):
@@ -487,7 +491,7 @@ def _parse_filters(filters: Union[str, dict[str, _Level]]) -> dict[str, _Level]:
         return {}
 
 
-def _parse_dict_of_filters(filters: dict[str, _Level]):
+def _parse_dict_of_filters(filters: Dict[str, _Level]) -> Dict[str, _Level]:
     for entry, level_name in filters.items():
         if _get_level_value(level_name) is None:
             logging.warning("Invalid filter entry [%s]", entry)
@@ -495,8 +499,8 @@ def _parse_dict_of_filters(filters: dict[str, _Level]):
     return filters
 
 
-def _parse_str_of_filters(filters: str):
-    chan_map: dict[str, _Level] = {}
+def _parse_str_of_filters(filters: str) -> Dict[str, _Level]:
+    chan_map: Dict[str, _Level] = {}
     for entry in filters.split(","):
         if len(entry):
             parts = entry.split(":")
@@ -531,11 +535,11 @@ def _parse_str_of_filters(filters: str):
 
 def configure(
     default_level: str,
-    filters: Union[str, dict[str, _Level]] = "",
+    filters: Union[str, Dict[str, _Level]] = "",
     formatter: Union[str, AlogFormatterBase] = "pretty",
     thread_id=False,
     handler_generator: Optional[Callable[[], logging.Handler]] = None,
-):
+) -> None:
     """Top-level configuration function for the alog module. This function
     configures the logging package to use the given default level and
     overwrites the levels for all filters as specified. It can also configure
@@ -637,7 +641,7 @@ def configure(
     g_filtered_channels = list(parsed_filters.keys())
 
 
-def use_channel(channel: Optional[str]):
+def use_channel(channel: Optional[str]) -> logging.Logger:
     """Interface wrapper for python alog implementation to keep consistency with
     other languages.
     """
@@ -655,7 +659,7 @@ class _ScopedLogBase:
     appropriate.
     """
 
-    def __init__(self, log_fn, format_str="", *args):
+    def __init__(self, log_fn, format_str="", *args) -> None:
         """Construct a new scoped logger."""
         self.log_fn = log_fn
         self.format_str = format_str
@@ -674,7 +678,7 @@ def __init__(self, log_fn, format_str="", *args):
         ), "Cannot use non-logging function for scoped log"
         self.enabled = self.log_fn.__self__.isEnabledFor(self.log_fn._level_value)
 
-    def _start_scoped_log(self):
+    def _start_scoped_log(self) -> None:
         """Log the start message for a scoped logger and increment the indentor."""
         if self.enabled:
             self.log_fn(scope_start_str + str(self.format_str), *self.args)
@@ -682,7 +686,7 @@ def _start_scoped_log(self):
             if g_alog_formatter:
                 g_alog_formatter.indent()
 
-    def _end_scoped_log(self):
+    def _end_scoped_log(self) -> None:
         """Log the end message for a scoped logger and decrement the indentor."""
         if self.enabled:
             global g_alog_formatter
@@ -702,12 +706,12 @@ class ScopedLog(_ScopedLogBase):
         >>>     _ = alog.ScopedLog(log_channel.debug)
     """
 
-    def __init__(self, log_fn, format_str="", *args):
+    def __init__(self, log_fn, format_str="", *args) -> None:
         """Construct a new scoped logger and print the begin message."""
         super().__init__(log_fn, format_str, *args)
         self._start_scoped_log()
 
-    def __del__(self):
+    def __del__(self) -> None:
         """Print the end message when this logger is deleted."""
         self._end_scoped_log()
 
@@ -724,12 +728,12 @@ class ContextLog(_ScopedLogBase):
         >>>   # logs the end message when the context manager exits
     """
 
-    def __enter__(self):
+    def __enter__(self) -> "ContextLog":
         """Log the begin message when the context manager starts."""
         self._start_scoped_log()
         return self
 
-    def __exit__(self, exception_type, exception_value, traceback):
+    def __exit__(self, exception_type, exception_value, traceback) -> None:
         """Log the end message when the context manager exits."""
         self._end_scoped_log()
 
@@ -751,7 +755,7 @@ class FunctionLog(ScopedLog):
         >>>     _ = alog.FunctionLog(log_channel.debug)
     """
 
-    def __init__(self, log_fn, format_str="", *args):
+    def __init__(self, log_fn, format_str="", *args) -> None:
         fn_name = (
             traceback.format_stack()[-2].strip().split(",")[2].split(" ")[2].strip()
         )
@@ -800,18 +804,18 @@ class _TimedLogBase:
     appropriate.
     """
 
-    def __init__(self, log_fn, format_str="", *args):
+    def __init__(self, log_fn, format_str="", *args) -> None:
         """Construct a new timed logger."""
         self.log_fn = log_fn
         self.format_str = format_str
         self.args = args
         self.start_time = 0
 
-    def _start_timed_log(self):
+    def _start_timed_log(self) -> None:
         """Get the start time for this timed logger."""
         self.start_time = time.time()
 
-    def _end_timed_log(self):
+    def _end_timed_log(self) -> None:
         """Gets the end time and prints the end message for this timed logger."""
         duration = timedelta(seconds=time.time() - self.start_time)
         fmt = self.format_str + "%s"
@@ -834,12 +838,12 @@ class ScopedTimer(_TimedLogBase):
         >>>     _ = alog.ScopedTimer(log_channel.debug)
     """
 
-    def __init__(self, log_fn, format_str="", *args):
+    def __init__(self, log_fn, format_str="", *args) -> None:
         """Construct a new scoped timer and get the start time."""
         super().__init__(log_fn, format_str, *args)
         self._start_timed_log()
 
-    def __del__(self):
+    def __del__(self) -> None:
         """Log the end message, including time delta, when this timer is deleted."""
         self._end_timed_log()
 
@@ -855,12 +859,12 @@ class ContextTimer(_TimedLogBase):
         >>>   # logs the time delta when the context manager exits
     """
 
-    def __enter__(self):
+    def __enter__(self) -> "ContextTimer":
         """Start the timer when a context is entered."""
         self._start_timed_log()
         return self
 
-    def __exit__(self, exception_type, exception_value, traceback):
+    def __exit__(self, exception_type, exception_value, traceback) -> None:
         """Log the end message, including time delta, when the context exits."""
         self._end_timed_log()
 

From 1967582f0c235cac07020604d7382bd785002774 Mon Sep 17 00:00:00 2001
From: Mynhardt Burger <mynhardt@gmail.com>
Date: Tue, 30 Jan 2024 09:11:07 -0500
Subject: [PATCH 5/7] type[] to Type[]

Signed-off-by: Mynhardt Burger <mynhardt@gmail.com>
---
 src/python/alog/alog.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/src/python/alog/alog.py b/src/python/alog/alog.py
index 83f6282..f192003 100644
--- a/src/python/alog/alog.py
+++ b/src/python/alog/alog.py
@@ -34,7 +34,7 @@
 import time
 import traceback
 from datetime import datetime, timedelta
-from typing import Any, Callable, Dict, List, Optional, Union
+from typing import Any, Callable, Dict, List, Optional, Type, Union
 
 _Level = Union[int, str]
 
@@ -328,7 +328,7 @@ def format(self, record: logging.LogRecord) -> str:
 }
 
 # Global map of default formatters
-g_alog_formatters: Dict[str, type[AlogFormatterBase]] = {
+g_alog_formatters: Dict[str, Type[AlogFormatterBase]] = {
     "json": AlogJsonFormatter,
     "pretty": AlogPrettyFormatter,
 }

From 8f31e194e2de3f1d8dde53137d41daa75616e782 Mon Sep 17 00:00:00 2001
From: Gabe Goodhart <gabe.l.hart@gmail.com>
Date: Tue, 30 Jan 2024 11:03:45 -0700
Subject: [PATCH 6/7] Style fix for slice indexing

Signed-off-by: Gabe Goodhart <gabe.l.hart@gmail.com>
---
 src/python/alog/alog.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/python/alog/alog.py b/src/python/alog/alog.py
index f192003..d5ed120 100644
--- a/src/python/alog/alog.py
+++ b/src/python/alog/alog.py
@@ -224,7 +224,7 @@ def _make_header(
         # Get the padded or truncated channel
         chan = channel
         if len(channel) > self.channel_len:
-            chan = channel[: self.channel_len]
+            chan = channel[:self.channel_len]
 
         elif len(channel) < self.channel_len:
             chan = channel + " " * (self.channel_len - len(channel))

From 3e66b252367862f4869728bb88d6d79af6a309db Mon Sep 17 00:00:00 2001
From: Gabe Goodhart <gabe.l.hart@gmail.com>
Date: Tue, 30 Jan 2024 11:04:06 -0700
Subject: [PATCH 7/7] Any type hint on __eq__

Signed-off-by: Gabe Goodhart <gabe.l.hart@gmail.com>
---
 src/python/alog/alog.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/python/alog/alog.py b/src/python/alog/alog.py
index d5ed120..3876ac7 100644
--- a/src/python/alog/alog.py
+++ b/src/python/alog/alog.py
@@ -356,7 +356,7 @@ class _MultiEqualString:
     def __init__(self, *strings: str) -> None:
         self._strings = strings
 
-    def __eq__(self, other) -> bool:
+    def __eq__(self, other: Any) -> bool:
         return other in self._strings