1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
|
import sentry_sdk
from sentry_sdk.utils import (
event_from_exception,
ensure_integration_enabled,
parse_version,
)
from sentry_sdk.integrations import DidNotEnable, Integration
from sentry_sdk.scope import should_send_default_pii
try:
import gql # type: ignore[import-not-found]
from graphql import print_ast, get_operation_ast, DocumentNode, VariableDefinitionNode # type: ignore[import-not-found]
from gql.transport import Transport, AsyncTransport # type: ignore[import-not-found]
from gql.transport.exceptions import TransportQueryError # type: ignore[import-not-found]
except ImportError:
raise DidNotEnable("gql is not installed")
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from typing import Any, Dict, Tuple, Union
from sentry_sdk._types import Event, EventProcessor
EventDataType = Dict[str, Union[str, Tuple[VariableDefinitionNode, ...]]]
MIN_GQL_VERSION = (3, 4, 1)
class GQLIntegration(Integration):
identifier = "gql"
@staticmethod
def setup_once():
# type: () -> None
gql_version = parse_version(gql.__version__)
if gql_version is None or gql_version < MIN_GQL_VERSION:
raise DidNotEnable(
"GQLIntegration is only supported for GQL versions %s and above."
% ".".join(str(num) for num in MIN_GQL_VERSION)
)
_patch_execute()
def _data_from_document(document):
# type: (DocumentNode) -> EventDataType
try:
operation_ast = get_operation_ast(document)
data = {"query": print_ast(document)} # type: EventDataType
if operation_ast is not None:
data["variables"] = operation_ast.variable_definitions
if operation_ast.name is not None:
data["operationName"] = operation_ast.name.value
return data
except (AttributeError, TypeError):
return dict()
def _transport_method(transport):
# type: (Union[Transport, AsyncTransport]) -> str
"""
The RequestsHTTPTransport allows defining the HTTP method; all
other transports use POST.
"""
try:
return transport.method
except AttributeError:
return "POST"
def _request_info_from_transport(transport):
# type: (Union[Transport, AsyncTransport, None]) -> Dict[str, str]
if transport is None:
return {}
request_info = {
"method": _transport_method(transport),
}
try:
request_info["url"] = transport.url
except AttributeError:
pass
return request_info
def _patch_execute():
# type: () -> None
real_execute = gql.Client.execute
@ensure_integration_enabled(GQLIntegration, real_execute)
def sentry_patched_execute(self, document, *args, **kwargs):
# type: (gql.Client, DocumentNode, Any, Any) -> Any
scope = sentry_sdk.get_isolation_scope()
scope.add_event_processor(_make_gql_event_processor(self, document))
try:
return real_execute(self, document, *args, **kwargs)
except TransportQueryError as e:
event, hint = event_from_exception(
e,
client_options=sentry_sdk.get_client().options,
mechanism={"type": "gql", "handled": False},
)
sentry_sdk.capture_event(event, hint)
raise e
gql.Client.execute = sentry_patched_execute
def _make_gql_event_processor(client, document):
# type: (gql.Client, DocumentNode) -> EventProcessor
def processor(event, hint):
# type: (Event, dict[str, Any]) -> Event
try:
errors = hint["exc_info"][1].errors
except (AttributeError, KeyError):
errors = None
request = event.setdefault("request", {})
request.update(
{
"api_target": "graphql",
**_request_info_from_transport(client.transport),
}
)
if should_send_default_pii():
request["data"] = _data_from_document(document)
contexts = event.setdefault("contexts", {})
response = contexts.setdefault("response", {})
response.update(
{
"data": {"errors": errors},
"type": response,
}
)
return event
return processor
|
