ref: Event Type (#2753)

Implements type hinting for Event via a TypedDict. This commit mainly adjusts type hints; however, there are also some minor code changes to make the code type-safe following the new changes.

Some items in the Event could have their types expanded by being defined as TypedDicts themselves. These items have been indicated with TODO comments.

Fixes GH-2357

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