11import logging
22import sys
3- from typing import List , Optional , Tuple , Union
43
54from sentry_sdk import capture_event
65from sentry_sdk .integrations .logging import ignore_logger
76from sentry_sdk .utils import event_from_exception
87
98
10- class SentryProcessor :
9+ class SentryProcessor ( object ) :
1110 """Sentry processor for structlog.
1211
1312 Uses Sentry SDK to capture events in Sentry.
1413 """
1514
1615 def __init__ (
17- self ,
18- level : int = logging .WARNING ,
19- active : bool = True ,
20- as_extra : bool = True ,
21- tag_keys : Union [List [str ], str ] = None ,
22- ) -> None :
16+ self , level = logging .WARNING , active = True , as_extra = True , tag_keys = None
17+ ):
18+ # type: (int, bool, bool, typing.Union[typing.List[str], str]) -> None
2319 """
2420 :param level: events of this or higher levels will be reported to Sentry.
2521 :param active: a flag to make this processor enabled/disabled.
@@ -34,7 +30,8 @@ def __init__(
3430 self ._as_extra = as_extra
3531 self ._original_event_dict = None
3632
37- def _get_event_and_hint (self , event_dict : dict ) -> Tuple [dict , Optional [str ]]:
33+ def _get_event_and_hint (self , event_dict ):
34+ # type: (dict) -> typing.Tuple[dict, typing.Optional[str]]
3835 """Create a sentry event and hint from structlog `event_dict` and sys.exc_info.
3936
4037 :param event_dict: structlog event_dict
@@ -61,15 +58,17 @@ def _get_event_and_hint(self, event_dict: dict) -> Tuple[dict, Optional[str]]:
6158
6259 return event , hint
6360
64- def _log (self , event_dict : dict ) -> str :
61+ def _log (self , event_dict ):
62+ # type: (dict) -> str
6563 """Send an event to Sentry and return sentry event id.
6664
6765 :param event_dict: structlog event_dict
6866 """
6967 event , hint = self ._get_event_and_hint (event_dict )
7068 return capture_event (event , hint = hint )
7169
72- def __call__ (self , logger , method , event_dict ) -> dict :
70+ def __call__ (self , logger , method , event_dict ):
71+ # type: (typing.Any, str, dict) -> dict
7372 """A middleware to process structlog `event_dict` and send it to Sentry."""
7473 self ._original_event_dict = event_dict .copy ()
7574 sentry_skip = event_dict .pop ("sentry_skip" , False )
@@ -92,16 +91,18 @@ class SentryJsonProcessor(SentryProcessor):
9291 Uses Sentry SDK to capture events in Sentry.
9392 """
9493
95- def __init__ (self , * args , ** kwargs ) -> None :
96- super ().__init__ (* args , ** kwargs )
94+ def __init__ (self , * args , ** kwargs ):
95+ super (SentryJsonProcessor , self ).__init__ (* args , ** kwargs )
9796 self ._is_logger_ignored = False
9897
99- def __call__ (self , logger , method , event_dict ) -> dict :
98+ def __call__ (self , logger , method , event_dict ):
99+ # type: (typing.Any, str, dict) -> dict
100100 if not self ._is_logger_ignored :
101101 self ._ignore_logger (logger , event_dict )
102- return super ().__call__ (logger , method , event_dict )
102+ return super (SentryJsonProcessor , self ).__call__ (logger , method , event_dict )
103103
104- def _ignore_logger (self , logger , event_dict : dict ) -> None :
104+ def _ignore_logger (self , logger , event_dict ):
105+ # type: (typing.Any, dict) -> None
105106 """Tell Sentry to ignore logger.
106107
107108 This is temporary workaround to prevent duplication of a JSON event in Sentry.
0 commit comments