Update logging configuration

core/service/logging/configuration.py

@@ -1,4 +1,8 @@
-from enum import Enum
+from __future__ import annotations
+
+import logging
+import sys
+from enum import auto
 from typing import Any
 
 import boto3
@@ -7,12 +11,60 @@
 
 from core.service.configuration import ServiceConfiguration
 
+# TODO: Remove this when we drop support for Python 3.10
+if sys.version_info >= (3, 11):
+    from enum import StrEnum
+else:
+    from backports.strenum import StrEnum
+
+
+class LogLevel(StrEnum):
+    """
+    A helper class to represent log levels as an Enum.
+
+    Since the logging module uses strings to represent log levels, the members of
+    this enum can be passed directly to the logging module to set the log level.
+    """
+
+    @staticmethod
+    def _generate_next_value_(
+        name: str, start: int, count: int, last_values: list[str]
+    ) -> str:
+        """
+        Return the upper-cased version of the member name.
+
+        By default, StrEnum uses the lower-cased version of the member name as the value,
+        but to match the logging module, we want to use the upper-cased version, so
+        we override this method to make auto() generate the correct value.
+        """
+        return name.upper()
+
+    debug = auto()
+    info = auto()
+    warning = auto()
+    error = auto()
+
+    @property
+    def levelno(self) -> int:
+        """
+        Return the integer value used by the logging module for this log level.
+        """
+        return logging._nameToLevel[self.value]
+
+    @classmethod
+    def from_level(cls, level: int | str) -> LogLevel:
+        """
+        Get a member of this enum from a string or integer log level.
+        """
+        if isinstance(level, int):
+            parsed_level = logging.getLevelName(level)
+        else:
+            parsed_level = str(level).upper()
 
-class LogLevel(Enum):
-    debug = "DEBUG"
-    info = "INFO"
-    warning = "WARNING"
-    error = "ERROR"
+        try:
+            return cls(parsed_level)
+        except ValueError:
+            raise ValueError(f"'{level}' is not a valid LogLevel") from None
 
 
 class LoggingConfiguration(ServiceConfiguration):

core/service/logging/container.py

@@ -35,7 +35,6 @@ class Logging(DeclarativeContainer):
     cloudwatch_handler: Provider[Handler | None] = providers.Singleton(
         create_cloudwatch_handler,
         formatter=json_formatter,
-        level=config.level,
         client=cloudwatch_client,
         group=config.cloudwatch_group,
         stream=config.cloudwatch_stream,
@@ -44,7 +43,7 @@ class Logging(DeclarativeContainer):
     )
 
     stream_handler: Provider[Handler] = providers.Singleton(
-        create_stream_handler, formatter=json_formatter, level=config.level
+        create_stream_handler, formatter=json_formatter
     )
 
     logging = providers.Resource(

core/service/logging/log.py

@@ -116,7 +116,6 @@ def filter(self, record: logging.LogRecord) -> bool:
 
 def create_cloudwatch_handler(
     formatter: logging.Formatter,
-    level: LogLevel,
     client: CloudWatchLogsClient,
     group: str,
     stream: str,
@@ -133,16 +132,12 @@ def create_cloudwatch_handler(
 
     handler.addFilter(LogLoopPreventionFilter())
     handler.setFormatter(formatter)
-    handler.setLevel(level.value)
     return handler
 
 
-def create_stream_handler(
-    formatter: logging.Formatter, level: LogLevel
-) -> logging.Handler:
+def create_stream_handler(formatter: logging.Formatter) -> logging.Handler:
     stream_handler = logging.StreamHandler()
     stream_handler.setFormatter(formatter)
-    stream_handler.setLevel(level.value)
     return stream_handler
 
 

pyproject.toml

@@ -181,6 +181,9 @@ version = "0"  # Version number is managed with tags in git
 [tool.poetry.dependencies]
 alembic = {extras = ["tz"], version = "^1.8.1"}
 aws-xray-sdk = "~2.13"
+# Lets us use StrEnum for Python 3.10 and below
+# TODO: Remove this when we drop support for Python 3.10
+"backports.strenum" = {version = "^1.3.1", python = "<3.11"}
 bcrypt = "^4.0.1"
 boto3 = "^1.28"
 certifi = "*"

tests/core/service/logging/test_configuration.py

@@ -1,7 +1,7 @@
 import pytest
 
 from core.config import CannotLoadConfiguration
-from core.service.logging.configuration import LoggingConfiguration
+from core.service.logging.configuration import LoggingConfiguration, LogLevel
 
 
 def test_cloudwatch_region_none() -> None:
@@ -30,3 +30,48 @@ def test_cloudwatch_region_valid() -> None:
     )
     assert config.cloudwatch_region == "us-east-2"
     assert config.cloudwatch_enabled is True
+
+
+class TestLogLevel:
+    def test_level_string(self) -> None:
+        assert LogLevel.debug == "DEBUG"
+        assert LogLevel.info == "INFO"  # type: ignore[unreachable]
+        assert LogLevel.warning == "WARNING"
+        assert LogLevel.error == "ERROR"
+
+    def test_levelno(self) -> None:
+        assert LogLevel.debug.levelno == 10
+        assert LogLevel.info.levelno == 20
+        assert LogLevel.warning.levelno == 30
+        assert LogLevel.error.levelno == 40
+
+    @pytest.mark.parametrize(
+        "level, expected",
+        [
+            (10, LogLevel.debug),
+            ("debug", LogLevel.debug),
+            ("DEBUG", LogLevel.debug),
+            ("info", LogLevel.info),
+            ("INFO", LogLevel.info),
+            (20, LogLevel.info),
+        ],
+    )
+    def test_from_level(self, level: int | str, expected: LogLevel) -> None:
+        assert LogLevel.from_level(level) == expected
+
+    @pytest.mark.parametrize(
+        "level",
+        [
+            "invalid",
+            "INVALID",
+            999,
+            41,
+            -1,
+            None,
+        ],
+    )
+    def test_from_level_invalid(self, level: str | int | None) -> None:
+        with pytest.raises(ValueError) as execinfo:
+            LogLevel.from_level(level)  # type: ignore[arg-type]
+
+        assert f"'{level}' is not a valid LogLevel" in str(execinfo.value)

tests/core/service/logging/test_log.py

@@ -209,7 +209,6 @@ def test_create_cloudwatch_handler() -> None:
 
     handler = create_cloudwatch_handler(
         formatter=mock_formatter,
-        level=LogLevel.info,
         client=mock_client,
         group="test_group",
         stream="test_stream",
@@ -223,18 +222,18 @@ def test_create_cloudwatch_handler() -> None:
     assert handler.send_interval == 13
     assert any(isinstance(f, LogLoopPreventionFilter) for f in handler.filters)
     assert handler.formatter == mock_formatter
-    assert handler.level == logging.INFO
+    assert handler.level == logging.NOTSET
 
 
 def test_create_stream_handler() -> None:
     mock_formatter = MagicMock()
 
-    handler = create_stream_handler(formatter=mock_formatter, level=LogLevel.debug)
+    handler = create_stream_handler(formatter=mock_formatter)
 
     assert isinstance(handler, logging.StreamHandler)
     assert not any(isinstance(f, LogLoopPreventionFilter) for f in handler.filters)
     assert handler.formatter == mock_formatter
-    assert handler.level == logging.DEBUG
+    assert handler.level == logging.NOTSET
 
 
 def test_setup_logging_cloudwatch_disabled() -> None:

tests/core/test_app_server.py

@@ -556,7 +556,7 @@ def test_handle_error_as_problem_detail_document(
         # The exception was logged at the error level
         assert len(caplog.records) == 1
         log_record = caplog.records[0]
-        assert log_record.levelname == LogLevel.error.value
+        assert log_record.levelname == LogLevel.error
         assert "Exception in web app" in log_record.message
 
     def test_handle_error_problem_error(
@@ -581,7 +581,7 @@ def test_handle_error_problem_error(
         # detail document had a status code of 502.
         assert len(caplog.records) == 1
         log_record = caplog.records[0]
-        assert log_record.levelname == LogLevel.warning.value
+        assert log_record.levelname == LogLevel.warning
         assert "Exception in web app" in log_record.message
 
     def test_handle_operational_error(
@@ -599,7 +599,7 @@ def test_handle_operational_error(
 
         assert len(caplog.records) == 1
         log_record = caplog.records[0]
-        assert log_record.levelname == LogLevel.warning.value
+        assert log_record.levelname == LogLevel.warning
 
 
 class TestCompressibleAnnotator:

tests/core/util/test_log.py

@@ -19,34 +19,34 @@ def test_method_2(self):
 
 
 def test_log_elapsed_time_cls(caplog: LogCaptureFixture):
-    caplog.set_level(LogLevel.info.value)
+    caplog.set_level(LogLevel.info)
 
     MockClass.test_method()
     assert len(caplog.records) == 2
 
     [first, second] = caplog.records
     assert first.name == "tests.core.util.test_log.MockClass"
     assert first.message == "Test: Starting..."
-    assert first.levelname == LogLevel.info.value
+    assert first.levelname == LogLevel.info
 
     assert second.name == "tests.core.util.test_log.MockClass"
     assert "Test: Completed. (elapsed time:" in second.message
-    assert second.levelname == LogLevel.info.value
+    assert second.levelname == LogLevel.info
 
 
 def test_log_elapsed_time_instance(caplog: LogCaptureFixture):
-    caplog.set_level(LogLevel.debug.value)
+    caplog.set_level(LogLevel.debug)
 
     MockClass().test_method_2()
     assert len(caplog.records) == 1
     [record] = caplog.records
     assert record.name == "tests.core.util.test_log.MockClass"
     assert "Test 12345: Completed. (elapsed time:" in record.message
-    assert record.levelname == LogLevel.debug.value
+    assert record.levelname == LogLevel.debug
 
 
 def test_log_elapsed_time_invalid(caplog: LogCaptureFixture):
-    caplog.set_level(LogLevel.info.value)
+    caplog.set_level(LogLevel.info)
 
     with pytest.raises(RuntimeError):
         log_elapsed_time(log_level=LogLevel.info, message_prefix="Test")(lambda: None)()