Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/sentry_sdk/integrations/atexit.py: 55%
36 statements
« prev ^ index » next coverage.py v6.4.4, created at 2023-07-17 14:22 -0600
« prev ^ index » next coverage.py v6.4.4, created at 2023-07-17 14:22 -0600
1from __future__ import absolute_import
3import os
4import sys
5import atexit
7from sentry_sdk.hub import Hub
8from sentry_sdk.utils import logger
9from sentry_sdk.integrations import Integration
11from sentry_sdk._types import MYPY
13if MYPY: 13 ↛ 15line 13 didn't jump to line 15, because the condition on line 13 was never true
15 from typing import Any
16 from typing import Optional
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 """
26 def echo(msg):
27 # type: (str) -> None
28 sys.stderr.write(msg + "\n")
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()
36class AtexitIntegration(Integration):
37 identifier = "atexit"
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
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")
57 # If there is a session on the hub, close it now.
58 hub.end_session()
60 # If an integration is there, a client has to be there.
61 client = hub.client # type: Any
62 client.close(callback=integration.callback)