feat: Add support for custom log levels

Author: radiophysicist

structlog_sentry/__init__.py

@@ -177,6 +177,21 @@ def _handle_breadcrumb(self, event_dict: EventDict) -> None:
             event, hint = self._get_breadcrumb_and_hint(event_dict)
             self._get_hub().add_breadcrumb(event, hint=hint)
 
+    @staticmethod
+    def _get_level_value(level_name: str) -> int:
+        """Get numeric value for the log level name given."""
+        try:
+            # Try to get one of predefined log levels
+            return getattr(logging, level_name)
+        except AttributeError as e:
+            # May be it is a custom log level?
+            level = logging.getLevelName(level_name)
+            if isinstance(level, int):
+                return level
+
+            # Re-raise original error
+            raise ValueError(f"{level_name} is not a valid log level") from e
+
     def __call__(
         self, logger: WrappedLogger, name: str, event_dict: EventDict
     ) -> EventDict:
@@ -185,7 +200,7 @@ def __call__(
         sentry_skip = event_dict.pop("sentry_skip", False)
 
         if self.active and not sentry_skip and self._can_record(logger, event_dict):
-            level = getattr(logging, event_dict["level"].upper())
+            level = self._get_level_value(event_dict["level"].upper())
 
             if level >= self.event_level:
                 self._handle_event(event_dict)

test/test_sentry_processor.py

@@ -11,6 +11,12 @@
 ]
 
 
+# Register custom log level
+CUSTOM_LOG_LEVEL_NAME = "CUSTOM_LEVEL"
+CUSTOM_LOG_LEVEL_VALUE = logging.DEBUG
+logging.addLevelName(CUSTOM_LOG_LEVEL_VALUE, CUSTOM_LOG_LEVEL_NAME)
+
+
 @dataclass
 class ClientParams:
     include_local_variables: bool = True
@@ -78,11 +84,19 @@ def test_sentry_sent():
     assert event_dict.get("sentry") == "sent"
 
 
-@pytest.mark.parametrize("level", ["debug", "info", "warning"])
-def test_sentry_log(sentry_events, level):
+@pytest.mark.parametrize(
+    "level, level_value",
+    [
+        (CUSTOM_LOG_LEVEL_NAME, CUSTOM_LOG_LEVEL_VALUE),
+        ("debug", logging.DEBUG),
+        ("info", logging.INFO),
+        ("warning", logging.WARNING),
+    ],
+)
+def test_sentry_log(sentry_events, level, level_value):
     event_data = {"level": level, "event": level + " message"}
 
-    processor = SentryProcessor(event_level=getattr(logging, level.upper()))
+    processor = SentryProcessor(event_level=level_value)
     processor(None, None, event_data)
 
     assert_event_dict(event_data, sentry_events)