Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/sentry_sdk/integrations/threading.py: 37%

1from __future__ import absolute_import 

2 

3import sys 

4from threading import Thread, current_thread 

5 

6from sentry_sdk import Hub 

7from sentry_sdk._compat import reraise 

8from sentry_sdk._types import MYPY 

9from sentry_sdk.integrations import Integration 

10from sentry_sdk.utils import event_from_exception, capture_internal_exceptions 

11 

12if MYPY: 12 ↛ 13line 12 didn't jump to line 13, because the condition on line 12 was never true

13 from typing import Any 

14 from typing import TypeVar 

15 from typing import Callable 

16 from typing import Optional 

17 

18 from sentry_sdk._types import ExcInfo 

19 

20 F = TypeVar("F", bound=Callable[..., Any]) 

21 

22 

23class ThreadingIntegration(Integration): 

24 identifier = "threading" 

25 

26 def __init__(self, propagate_hub=False): 

27 # type: (bool) -> None 

28 self.propagate_hub = propagate_hub 

29 

30 @staticmethod 

31 def setup_once(): 

32 # type: () -> None 

33 old_start = Thread.start 

34 

35 def sentry_start(self, *a, **kw): 

36 # type: (Thread, *Any, **Any) -> Any 

37 hub = Hub.current 

38 integration = hub.get_integration(ThreadingIntegration) 

39 if integration is not None: 

40 if not integration.propagate_hub: 

41 hub_ = None 

42 else: 

43 hub_ = Hub(hub) 

44 # Patching instance methods in `start()` creates a reference cycle if 

45 # done in a naive way. See 

46 # https://github.com/getsentry/sentry-python/pull/434 

47 # 

48 # In threading module, using current_thread API will access current thread instance 

49 # without holding it to avoid a reference cycle in an easier way. 

50 with capture_internal_exceptions(): 

51 new_run = _wrap_run(hub_, getattr(self.run, "__func__", self.run)) 

52 self.run = new_run # type: ignore 

53 

54 return old_start(self, *a, **kw) 

55 

56 Thread.start = sentry_start # type: ignore 

57 

58 

59def _wrap_run(parent_hub, old_run_func): 

60 # type: (Optional[Hub], F) -> F 

61 def run(*a, **kw): 

62 # type: (*Any, **Any) -> Any 

63 hub = parent_hub or Hub.current 

64 with hub: 

65 try: 

66 self = current_thread() 

67 return old_run_func(self, *a, **kw) 

68 except Exception: 

69 reraise(*_capture_exception()) 

70 

71 return run # type: ignore 

72 

73 

74def _capture_exception(): 

75 # type: () -> ExcInfo 

76 hub = Hub.current 

77 exc_info = sys.exc_info() 

78 

79 if hub.get_integration(ThreadingIntegration) is not None: 

80 # If an integration is there, a client has to be there. 

81 client = hub.client # type: Any 

82 

83 event, hint = event_from_exception( 

84 exc_info, 

85 client_options=client.options, 

86 mechanism={"type": "threading", "handled": False}, 

87 ) 

88 hub.capture_event(event, hint=hint) 

89 

90 return exc_info