ref: Event Type

sentry_sdk/_types.py

@@ -9,6 +9,10 @@
 
 
 if TYPE_CHECKING:
+    from collections.abc import MutableMapping
+
+    from datetime import datetime
+
     from types import TracebackType
     from typing import Any
     from typing import Callable
@@ -19,13 +23,69 @@
     from typing import Tuple
     from typing import Type
     from typing import Union
-    from typing_extensions import Literal
+    from typing_extensions import Literal, TypedDict
+
+    # "critical" is an alias of "fatal" recognized by Relay
+    LogLevelStr = Literal["fatal", "critical", "error", "warning", "info", "debug"]
+
+    Event = TypedDict(
+        "Event",
+        {
+            "breadcrumbs": dict[
+                Literal["values"], list[dict[str, Any]]
+            ],  # TODO: We can expand on this type
+            "check_in_id": str,
+            "contexts": dict[str, dict[str, object]],
+            "dist": str,
+            "duration": Optional[float],
+            "environment": str,
+            "errors": list[dict[str, Any]],  # TODO: We can expand on this type
+            "event_id": str,
+            "exception": dict[
+                Literal["values"], list[dict[str, Any]]
+            ],  # TODO: We can expand on this type
+            "extra": MutableMapping[str, object],
+            "fingerprint": list[str],
+            "level": LogLevelStr,
+            "logentry": Mapping[str, object],
+            "logger": str,
+            "measurements": dict[str, object],
+            "message": str,
+            "modules": dict[str, str],
+            "monitor_config": Mapping[str, object],
+            "monitor_slug": Optional[str],
+            "platform": Literal["python"],
+            "profile": object,  # Should be sentry_sdk.profiler.Profile, but we can't import that here due to circular imports
+            "release": str,
+            "request": dict[str, object],
+            "sdk": Mapping[str, object],
+            "server_name": str,
+            "spans": list[dict[str, object]],
+            "stacktrace": dict[
+                str, object
+            ],  # We access this key in the code, but I am unsure whether we ever set it
+            "start_timestamp": datetime,
+            "status": Optional[str],
+            "tags": MutableMapping[
+                str, str
+            ],  # Tags must be less than 200 characters each
+            "threads": dict[
+                Literal["values"], list[dict[str, Any]]
+            ],  # TODO: We can expand on this type
+            "timestamp": Optional[datetime],  # Must be set before sending the event
+            "transaction": str,
+            "transaction_info": Mapping[str, Any],  # TODO: We can expand on this type
+            "type": Literal["check_in", "transaction"],
+            "user": dict[str, object],
+            "_metrics_summary": dict[str, object],
+        },
+        total=False,
+    )
 
     ExcInfo = Tuple[
         Optional[Type[BaseException]], Optional[BaseException], Optional[TracebackType]
     ]
 
-    Event = Dict[str, Any]
     Hint = Dict[str, Any]
 
     Breadcrumb = Dict[str, Any]

sentry_sdk/api.py

@@ -22,6 +22,7 @@
         BreadcrumbHint,
         ExcInfo,
         MeasurementUnit,
+        LogLevelStr,
     )
     from sentry_sdk.tracing import Span
 
@@ -91,7 +92,7 @@ def capture_event(
 @hubmethod
 def capture_message(
     message,  # type: str
-    level=None,  # type: Optional[str]
+    level=None,  # type: Optional[LogLevelStr]
     scope=None,  # type: Optional[Any]
     **scope_kwargs  # type: Any
 ):
@@ -189,7 +190,7 @@ def set_user(value):
 
 @scopemethod
 def set_level(value):
-    # type: (str) -> None
+    # type: (LogLevelStr) -> None
     return Hub.current.scope.set_level(value)
 
 

sentry_sdk/client.py

@@ -1,3 +1,8 @@
+try:
+    from collections.abc import Mapping
+except ImportError:
+    from collections import Mapping  # type: ignore[attr-defined]
+
 from importlib import import_module
 import os
 import uuid
@@ -38,7 +43,7 @@
 from sentry_sdk.utils import ContextVar
 from sentry_sdk.sessions import SessionFlusher
 from sentry_sdk.envelope import Envelope
-from sentry_sdk.profiler import has_profiling_enabled, setup_profiler
+from sentry_sdk.profiler import has_profiling_enabled, Profile, setup_profiler
 from sentry_sdk.scrubber import EventScrubber
 from sentry_sdk.monitor import Monitor
 from sentry_sdk.spotlight import setup_spotlight
@@ -393,7 +398,7 @@ def _prepare_event(
 
         for key in "release", "environment", "server_name", "dist":
             if event.get(key) is None and self.options[key] is not None:
-                event[key] = text_type(self.options[key]).strip()
+                event[key] = text_type(self.options[key]).strip()  # type: ignore[literal-required]
         if event.get("sdk") is None:
             sdk_info = dict(SDK_INFO)
             sdk_info["integrations"] = sorted(self.integrations.keys())
@@ -567,7 +572,7 @@ def _update_session_from_event(
             errored = True
             for error in exceptions:
                 mechanism = error.get("mechanism")
-                if mechanism and mechanism.get("handled") is False:
+                if isinstance(mechanism, Mapping) and mechanism.get("handled") is False:
                     crashed = True
                     break
 
@@ -659,15 +664,15 @@ def capture_event(
             headers = {
                 "event_id": event_opt["event_id"],
                 "sent_at": format_timestamp(datetime_utcnow()),
-            }
+            }  # type: dict[str, object]
 
             if dynamic_sampling_context:
                 headers["trace"] = dynamic_sampling_context
 
             envelope = Envelope(headers=headers)
 
             if is_transaction:
-                if profile is not None:
+                if isinstance(profile, Profile):
                     envelope.add_profile(profile.to_json(event_opt, self.options))
                 envelope.add_transaction(event_opt)
             elif is_checkin:

sentry_sdk/crons/api.py

@@ -6,6 +6,7 @@
 
 if TYPE_CHECKING:
     from typing import Any, Dict, Optional
+    from sentry_sdk._types import Event
 
 
 def _create_check_in_event(
@@ -15,7 +16,7 @@ def _create_check_in_event(
     duration_s=None,
     monitor_config=None,
 ):
-    # type: (Optional[str], Optional[str], Optional[str], Optional[float], Optional[Dict[str, Any]]) -> Dict[str, Any]
+    # type: (Optional[str], Optional[str], Optional[str], Optional[float], Optional[Dict[str, Any]]) -> Event
     options = Hub.current.client.options if Hub.current.client else {}
     check_in_id = check_in_id or uuid.uuid4().hex  # type: str
 
@@ -27,7 +28,7 @@ def _create_check_in_event(
         "duration": duration_s,
         "environment": options.get("environment", None),
         "release": options.get("release", None),
-    }
+    }  # type: Event
 
     if monitor_config:
         check_in["monitor_config"] = monitor_config

sentry_sdk/hub.py

@@ -40,6 +40,7 @@
         Breadcrumb,
         BreadcrumbHint,
         ExcInfo,
+        LogLevelStr,
     )
     from sentry_sdk.consts import ClientConstructor
 
@@ -335,7 +336,7 @@ def capture_event(self, event, hint=None, scope=None, **scope_kwargs):
         return last_event_id
 
     def capture_message(self, message, level=None, scope=None, **scope_kwargs):
-        # type: (str, Optional[str], Optional[Scope], Any) -> Optional[str]
+        # type: (str, Optional[LogLevelStr], Optional[Scope], Any) -> Optional[str]
         """
         Captures a message.
 

sentry_sdk/integrations/_wsgi_common.py

@@ -22,6 +22,7 @@
     from typing import Dict
     from typing import Optional
     from typing import Union
+    from sentry_sdk._types import Event
 
 
 SENSITIVE_ENV_KEYS = (
@@ -59,7 +60,7 @@ def __init__(self, request):
         self.request = request
 
     def extract_into_event(self, event):
-        # type: (Dict[str, Any]) -> None
+        # type: (Event) -> None
         client = Hub.current.client
         if client is None:
             return

sentry_sdk/integrations/aiohttp.py

@@ -48,13 +48,12 @@
     from aiohttp import TraceRequestStartParams, TraceRequestEndParams
     from types import SimpleNamespace
     from typing import Any
-    from typing import Dict
     from typing import Optional
     from typing import Tuple
     from typing import Union
 
     from sentry_sdk.utils import ExcInfo
-    from sentry_sdk._types import EventProcessor
+    from sentry_sdk._types import Event, EventProcessor
 
 
 TRANSACTION_STYLE_VALUES = ("handler_name", "method_and_path_pattern")
@@ -256,10 +255,10 @@ async def on_request_end(session, trace_config_ctx, params):
 def _make_request_processor(weak_request):
     # type: (weakref.ReferenceType[Request]) -> EventProcessor
     def aiohttp_processor(
-        event,  # type: Dict[str, Any]
-        hint,  # type: Dict[str, Tuple[type, BaseException, Any]]
+        event,  # type: Event
+        hint,  # type: dict[str, Tuple[type, BaseException, Any]]
     ):
-        # type: (...) -> Dict[str, Any]
+        # type: (...) -> Event
         request = weak_request()
         if request is None:
             return event

sentry_sdk/integrations/ariadne.py

@@ -23,7 +23,7 @@
     from typing import Any, Dict, List, Optional
     from ariadne.types import GraphQLError, GraphQLResult, GraphQLSchema, QueryParser  # type: ignore
     from graphql.language.ast import DocumentNode  # type: ignore
-    from sentry_sdk._types import EventProcessor
+    from sentry_sdk._types import Event, EventProcessor
 
 
 class AriadneIntegration(Integration):
@@ -131,7 +131,7 @@ def _make_request_event_processor(data):
     """Add request data and api_target to events."""
 
     def inner(event, hint):
-        # type: (Dict[str, Any], Dict[str, Any]) -> Dict[str, Any]
+        # type: (Event, dict[str, Any]) -> Event
         if not isinstance(data, dict):
             return event
 
@@ -163,7 +163,7 @@ def _make_response_event_processor(response):
     """Add response data to the event's response context."""
 
     def inner(event, hint):
-        # type: (Dict[str, Any], Dict[str, Any]) -> Dict[str, Any]
+        # type: (Event, dict[str, Any]) -> Event
         with capture_internal_exceptions():
             if _should_send_default_pii() and response.get("errors"):
                 contexts = event.setdefault("contexts", {})

sentry_sdk/integrations/bottle.py

@@ -200,7 +200,7 @@ def _make_request_event_processor(app, request, integration):
     # type: (Bottle, LocalRequest, BottleIntegration) -> EventProcessor
 
     def event_processor(event, hint):
-        # type: (Dict[str, Any], Dict[str, Any]) -> Dict[str, Any]
+        # type: (Event, dict[str, Any]) -> Event
         _set_transaction_name_and_source(event, integration.transaction_style, request)
 
         with capture_internal_exceptions():

sentry_sdk/integrations/django/__init__.py

@@ -472,7 +472,7 @@ def sentry_patched_get_response(self, request):
 def _make_wsgi_request_event_processor(weak_request, integration):
     # type: (Callable[[], WSGIRequest], DjangoIntegration) -> EventProcessor
     def wsgi_request_event_processor(event, hint):
-        # type: (Dict[str, Any], Dict[str, Any]) -> Dict[str, Any]
+        # type: (Event, dict[str, Any]) -> Event
         # if the request is gone we are fine not logging the data from
         # it.  This might happen if the processor is pushed away to
         # another thread.
@@ -570,7 +570,7 @@ def parsed_body(self):
 
 
 def _set_user_info(request, event):
-    # type: (WSGIRequest, Dict[str, Any]) -> None
+    # type: (WSGIRequest, Event) -> None
     user_info = event.setdefault("user", {})
 
     user = getattr(request, "user", None)

sentry_sdk/integrations/django/asgi.py

@@ -26,13 +26,13 @@
     from django.core.handlers.asgi import ASGIRequest
     from django.http.response import HttpResponse
 
-    from sentry_sdk._types import EventProcessor
+    from sentry_sdk._types import Event, EventProcessor
 
 
 def _make_asgi_request_event_processor(request):
     # type: (ASGIRequest) -> EventProcessor
     def asgi_request_event_processor(event, hint):
-        # type: (dict[str, Any], dict[str, Any]) -> dict[str, Any]
+        # type: (Event, dict[str, Any]) -> Event
         # if the request is gone we are fine not logging the data from
         # it.  This might happen if the processor is pushed away to
         # another thread.

sentry_sdk/integrations/falcon.py

@@ -18,7 +18,7 @@
     from typing import Dict
     from typing import Optional
 
-    from sentry_sdk._types import EventProcessor
+    from sentry_sdk._types import Event, EventProcessor
 
 # In Falcon 3.0 `falcon.api_helpers` is renamed to `falcon.app_helpers`
 # and `falcon.API` to `falcon.App`
@@ -258,7 +258,7 @@ def _has_http_5xx_status(response):
 
 
 def _set_transaction_name_and_source(event, transaction_style, request):
-    # type: (Dict[str, Any], str, falcon.Request) -> None
+    # type: (Event, str, falcon.Request) -> None
     name_for_style = {
         "uri_template": request.uri_template,
         "path": request.path,
@@ -271,7 +271,7 @@ def _make_request_event_processor(req, integration):
     # type: (falcon.Request, FalconIntegration) -> EventProcessor
 
     def event_processor(event, hint):
-        # type: (Dict[str, Any], Dict[str, Any]) -> Dict[str, Any]
+        # type: (Event, dict[str, Any]) -> Event
         _set_transaction_name_and_source(event, integration.transaction_style, req)
 
         with capture_internal_exceptions():

sentry_sdk/integrations/fastapi.py

@@ -11,6 +11,7 @@
 if TYPE_CHECKING:
     from typing import Any, Callable, Dict
     from sentry_sdk.scope import Scope
+    from sentry_sdk._types import Event
 
 try:
     from sentry_sdk.integrations.starlette import (
@@ -111,9 +112,9 @@ async def _sentry_app(*args, **kwargs):
                 info = await extractor.extract_request_info()
 
                 def _make_request_event_processor(req, integration):
-                    # type: (Any, Any) -> Callable[[Dict[str, Any], Dict[str, Any]], Dict[str, Any]]
+                    # type: (Any, Any) -> Callable[[Event, Dict[str, Any]], Event]
                     def event_processor(event, hint):
-                        # type: (Dict[str, Any], Dict[str, Any]) -> Dict[str, Any]
+                        # type: (Event, Dict[str, Any]) -> Event
 
                         # Extract information from request
                         request_info = event.get("request", {})

sentry_sdk/integrations/flask.py

@@ -16,7 +16,7 @@
 if TYPE_CHECKING:
     from typing import Any, Callable, Dict, Union
 
-    from sentry_sdk._types import EventProcessor
+    from sentry_sdk._types import Event, EventProcessor
     from sentry_sdk.integrations.wsgi import _ScopedResponse
     from werkzeug.datastructures import FileStorage, ImmutableMultiDict
 
@@ -172,7 +172,7 @@ def _make_request_event_processor(app, request, integration):
     # type: (Flask, Callable[[], Request], FlaskIntegration) -> EventProcessor
 
     def inner(event, hint):
-        # type: (Dict[str, Any], Dict[str, Any]) -> Dict[str, Any]
+        # type: (Event, dict[str, Any]) -> Event
 
         # if the request is gone we are fine not logging the data from
         # it.  This might happen if the processor is pushed away to
@@ -211,7 +211,7 @@ def _capture_exception(sender, exception, **kwargs):
 
 
 def _add_user_to_event(event):
-    # type: (Dict[str, Any]) -> None
+    # type: (Event) -> None
     if flask_login is None:
         return