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

1from __future__ import absolute_import 

2 

3from sentry_sdk._types import MYPY 

4from sentry_sdk.hub import Hub 

5from sentry_sdk.integrations import Integration, DidNotEnable 

6from sentry_sdk.tracing_utils import record_sql_queries 

7 

8try: 

9 from sqlalchemy.engine import Engine # type: ignore 

10 from sqlalchemy.event import listen # type: ignore 

11 from sqlalchemy import __version__ as SQLALCHEMY_VERSION # type: ignore 

12except ImportError: 

13 raise DidNotEnable("SQLAlchemy not installed.") 

14 

15if MYPY: 

16 from typing import Any 

17 from typing import ContextManager 

18 from typing import Optional 

19 

20 from sentry_sdk.tracing import Span 

21 

22 

23class SqlalchemyIntegration(Integration): 

24 identifier = "sqlalchemy" 

25 

26 @staticmethod 

27 def setup_once(): 

28 # type: () -> None 

29 

30 try: 

31 version = tuple(map(int, SQLALCHEMY_VERSION.split("b")[0].split("."))) 

32 except (TypeError, ValueError): 

33 raise DidNotEnable( 

34 "Unparsable SQLAlchemy version: {}".format(SQLALCHEMY_VERSION) 

35 ) 

36 

37 if version < (1, 2): 

38 raise DidNotEnable("SQLAlchemy 1.2 or newer required.") 

39 

40 listen(Engine, "before_cursor_execute", _before_cursor_execute) 

41 listen(Engine, "after_cursor_execute", _after_cursor_execute) 

42 listen(Engine, "handle_error", _handle_error) 

43 

44 

45def _before_cursor_execute( 

46 conn, cursor, statement, parameters, context, executemany, *args 

47): 

48 # type: (Any, Any, Any, Any, Any, bool, *Any) -> None 

49 hub = Hub.current 

50 if hub.get_integration(SqlalchemyIntegration) is None: 

51 return 

52 

53 ctx_mgr = record_sql_queries( 

54 hub, 

55 cursor, 

56 statement, 

57 parameters, 

58 paramstyle=context and context.dialect and context.dialect.paramstyle or None, 

59 executemany=executemany, 

60 ) 

61 context._sentry_sql_span_manager = ctx_mgr 

62 

63 span = ctx_mgr.__enter__() 

64 

65 if span is not None: 

66 context._sentry_sql_span = span 

67 

68 

69def _after_cursor_execute(conn, cursor, statement, parameters, context, *args): 

70 # type: (Any, Any, Any, Any, Any, *Any) -> None 

71 ctx_mgr = getattr( 

72 context, "_sentry_sql_span_manager", None 

73 ) # type: Optional[ContextManager[Any]] 

74 

75 if ctx_mgr is not None: 

76 context._sentry_sql_span_manager = None 

77 ctx_mgr.__exit__(None, None, None) 

78 

79 

80def _handle_error(context, *args): 

81 # type: (Any, *Any) -> None 

82 execution_context = context.execution_context 

83 if execution_context is None: 

84 return 

85 

86 span = getattr(execution_context, "_sentry_sql_span", None) # type: Optional[Span] 

87 

88 if span is not None: 

89 span.set_status("internal_error") 

90 

91 # _after_cursor_execute does not get called for crashing SQL stmts. Judging 

92 # from SQLAlchemy codebase it does seem like any error coming into this 

93 # handler is going to be fatal. 

94 ctx_mgr = getattr( 

95 execution_context, "_sentry_sql_span_manager", None 

96 ) # type: Optional[ContextManager[Any]] 

97 

98 if ctx_mgr is not None: 

99 execution_context._sentry_sql_span_manager = None 

100 ctx_mgr.__exit__(None, None, None)