Skip to content

Commit

Permalink
🏷️ Add/update type annotations
Browse files Browse the repository at this point in the history
Based on feedback from Mypy.
  • Loading branch information
TeoZosa committed Sep 12, 2021
1 parent ac3fe0d commit e019a56
Showing 1 changed file with 13 additions and 10 deletions.
23 changes: 13 additions & 10 deletions structlog_sentry/__init__.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import logging
import sys
from typing import Iterable, List, Optional, Set, Tuple, Union
from typing import Any, Dict, Iterable, List, Optional, Set, Tuple, Union

from sentry_sdk import capture_event
from sentry_sdk.integrations.logging import (
Expand Down Expand Up @@ -36,13 +36,13 @@ def __init__(
self.active = active
self.tag_keys = tag_keys
self._as_extra = as_extra
self._original_event_dict = None
self._original_event_dict: dict = None
self._ignored_loggers: Set[str] = set()
if ignore_loggers is not None:
self._ignored_loggers.update(set(ignore_loggers))

@staticmethod
def _get_logger_name(logger, event_dict: dict) -> Optional[str]:
def _get_logger_name(logger: Any, event_dict: dict) -> Optional[str]:
"""Get logger name from event_dict with a fallbacks to logger.name and record.name
:param logger: logger instance
Expand All @@ -62,7 +62,9 @@ def _get_logger_name(logger, event_dict: dict) -> Optional[str]:

return logger_name

def _get_event_and_hint(self, event_dict: dict) -> Tuple[dict, Optional[str]]:
def _get_event_and_hint(
self, event_dict: dict
) -> Tuple[dict, Optional[Dict[str, Any]]]:
"""Create a sentry event and hint from structlog `event_dict` and sys.exc_info.
:param event_dict: structlog event_dict
Expand All @@ -73,6 +75,7 @@ def _get_event_and_hint(self, event_dict: dict) -> Tuple[dict, Optional[str]]:
exc_info = sys.exc_info()
has_exc_info = exc_info and exc_info != (None, None, None)

hint: Optional[Dict[str, Any]]
if has_exc_info:
event, hint = event_from_exception(exc_info)
else:
Expand All @@ -94,15 +97,15 @@ def _get_event_and_hint(self, event_dict: dict) -> Tuple[dict, Optional[str]]:

return event, hint

def _log(self, event_dict: dict) -> str:
def _log(self, event_dict: dict) -> Optional[str]:
"""Send an event to Sentry and return sentry event id.
:param event_dict: structlog event_dict
"""
event, hint = self._get_event_and_hint(event_dict)
return capture_event(event, hint=hint)

def __call__(self, logger, method, event_dict) -> dict:
def __call__(self, logger: Any, method: Any, event_dict: dict) -> dict:
"""A middleware to process structlog `event_dict` and send it to Sentry."""
logger_name = self._get_logger_name(logger=logger, event_dict=event_dict)
if logger_name in self._ignored_loggers:
Expand Down Expand Up @@ -130,19 +133,19 @@ class SentryJsonProcessor(SentryProcessor):
Uses Sentry SDK to capture events in Sentry.
"""

def __init__(self, *args, **kwargs) -> None:
def __init__(self, *args: Any, **kwargs: Any) -> None:
super().__init__(*args, **kwargs)
# A set of all encountered structured logger names. If an application uses
# multiple loggers with different names (eg. different qualnames), then each of
# those loggers needs to be ignored in Sentry's logging integration so that this
# processor will be the only thing reporting the events.
self._ignored = set()
self._ignored: set = set()

def __call__(self, logger, method, event_dict) -> dict:
def __call__(self, logger: Any, method: Any, event_dict: dict) -> dict:
self._ignore_logger(logger, event_dict)
return super().__call__(logger, method, event_dict)

def _ignore_logger(self, logger, event_dict: dict) -> None:
def _ignore_logger(self, logger: Any, event_dict: dict) -> None:
"""Tell Sentry to ignore logger, if we haven't already.
This is temporary workaround to prevent duplication of a JSON event in Sentry.
Expand Down

0 comments on commit e019a56

Please sign in to comment.