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

1from __future__ import absolute_import 

2 

3import os 

4import sys 

5import atexit 

6 

7from sentry_sdk.hub import Hub 

8from sentry_sdk.utils import logger 

9from sentry_sdk.integrations import Integration 

10 

11from sentry_sdk._types import MYPY 

12 

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

14 

15 from typing import Any 

16 from typing import Optional 

17 

18 

19def default_callback(pending, timeout): 

20 # type: (int, int) -> None 

21 """This is the default shutdown callback that is set on the options. 

22 It prints out a message to stderr that informs the user that some events 

23 are still pending and the process is waiting for them to flush out. 

24 """ 

25 

26 def echo(msg): 

27 # type: (str) -> None 

28 sys.stderr.write(msg + "\n") 

29 

30 echo("Sentry is attempting to send %i pending error messages" % pending) 

31 echo("Waiting up to %s seconds" % timeout) 

32 echo("Press Ctrl-%s to quit" % (os.name == "nt" and "Break" or "C")) 

33 sys.stderr.flush() 

34 

35 

36class AtexitIntegration(Integration): 

37 identifier = "atexit" 

38 

39 def __init__(self, callback=None): 

40 # type: (Optional[Any]) -> None 

41 if callback is None: 41 ↛ 43line 41 didn't jump to line 43, because the condition on line 41 was never false

42 callback = default_callback 

43 self.callback = callback 

44 

45 @staticmethod 

46 def setup_once(): 

47 # type: () -> None 

48 @atexit.register 

49 def _shutdown(): 

50 # type: () -> None 

51 logger.debug("atexit: got shutdown signal") 

52 hub = Hub.main 

53 integration = hub.get_integration(AtexitIntegration) 

54 if integration is not None: 

55 logger.debug("atexit: shutting down client") 

56 

57 # If there is a session on the hub, close it now. 

58 hub.end_session() 

59 

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

61 client = hub.client # type: Any 

62 client.close(callback=integration.callback)