File size: 5,834 Bytes
9c6594c
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
from importlib import import_module

import sentry_sdk
from sentry_sdk import get_client, capture_event
from sentry_sdk.integrations import _check_minimum_version, DidNotEnable, Integration
from sentry_sdk.integrations.logging import ignore_logger
from sentry_sdk.integrations._wsgi_common import request_body_within_bounds
from sentry_sdk.scope import should_send_default_pii
from sentry_sdk.utils import (
    capture_internal_exceptions,
    ensure_integration_enabled,
    event_from_exception,
    package_version,
)

try:
    # importing like this is necessary due to name shadowing in ariadne
    # (ariadne.graphql is also a function)
    ariadne_graphql = import_module("ariadne.graphql")
except ImportError:
    raise DidNotEnable("ariadne is not installed")

from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from typing import Any, Dict, List, Optional
    from ariadne.types import GraphQLError, GraphQLResult, GraphQLSchema, QueryParser  # type: ignore
    from graphql.language.ast import DocumentNode
    from sentry_sdk._types import Event, EventProcessor


class AriadneIntegration(Integration):
    identifier = "ariadne"

    @staticmethod
    def setup_once():
        # type: () -> None
        version = package_version("ariadne")
        _check_minimum_version(AriadneIntegration, version)

        ignore_logger("ariadne")

        _patch_graphql()


def _patch_graphql():
    # type: () -> None
    old_parse_query = ariadne_graphql.parse_query
    old_handle_errors = ariadne_graphql.handle_graphql_errors
    old_handle_query_result = ariadne_graphql.handle_query_result

    @ensure_integration_enabled(AriadneIntegration, old_parse_query)
    def _sentry_patched_parse_query(context_value, query_parser, data):
        # type: (Optional[Any], Optional[QueryParser], Any) -> DocumentNode
        event_processor = _make_request_event_processor(data)
        sentry_sdk.get_isolation_scope().add_event_processor(event_processor)

        result = old_parse_query(context_value, query_parser, data)
        return result

    @ensure_integration_enabled(AriadneIntegration, old_handle_errors)
    def _sentry_patched_handle_graphql_errors(errors, *args, **kwargs):
        # type: (List[GraphQLError], Any, Any) -> GraphQLResult
        result = old_handle_errors(errors, *args, **kwargs)

        event_processor = _make_response_event_processor(result[1])
        sentry_sdk.get_isolation_scope().add_event_processor(event_processor)

        client = get_client()
        if client.is_active():
            with capture_internal_exceptions():
                for error in errors:
                    event, hint = event_from_exception(
                        error,
                        client_options=client.options,
                        mechanism={
                            "type": AriadneIntegration.identifier,
                            "handled": False,
                        },
                    )
                    capture_event(event, hint=hint)

        return result

    @ensure_integration_enabled(AriadneIntegration, old_handle_query_result)
    def _sentry_patched_handle_query_result(result, *args, **kwargs):
        # type: (Any, Any, Any) -> GraphQLResult
        query_result = old_handle_query_result(result, *args, **kwargs)

        event_processor = _make_response_event_processor(query_result[1])
        sentry_sdk.get_isolation_scope().add_event_processor(event_processor)

        client = get_client()
        if client.is_active():
            with capture_internal_exceptions():
                for error in result.errors or []:
                    event, hint = event_from_exception(
                        error,
                        client_options=client.options,
                        mechanism={
                            "type": AriadneIntegration.identifier,
                            "handled": False,
                        },
                    )
                    capture_event(event, hint=hint)

        return query_result

    ariadne_graphql.parse_query = _sentry_patched_parse_query  # type: ignore
    ariadne_graphql.handle_graphql_errors = _sentry_patched_handle_graphql_errors  # type: ignore
    ariadne_graphql.handle_query_result = _sentry_patched_handle_query_result  # type: ignore


def _make_request_event_processor(data):
    # type: (GraphQLSchema) -> EventProcessor
    """Add request data and api_target to events."""

    def inner(event, hint):
        # type: (Event, dict[str, Any]) -> Event
        if not isinstance(data, dict):
            return event

        with capture_internal_exceptions():
            try:
                content_length = int(
                    (data.get("headers") or {}).get("Content-Length", 0)
                )
            except (TypeError, ValueError):
                return event

            if should_send_default_pii() and request_body_within_bounds(
                get_client(), content_length
            ):
                request_info = event.setdefault("request", {})
                request_info["api_target"] = "graphql"
                request_info["data"] = data

            elif event.get("request", {}).get("data"):
                del event["request"]["data"]

        return event

    return inner


def _make_response_event_processor(response):
    # type: (Dict[str, Any]) -> EventProcessor
    """Add response data to the event's response context."""

    def inner(event, hint):
        # type: (Event, dict[str, Any]) -> Event
        with capture_internal_exceptions():
            if should_send_default_pii() and response.get("errors"):
                contexts = event.setdefault("contexts", {})
                contexts["response"] = {
                    "data": response,
                }

        return event

    return inner