format

Author: sl0thentr0py

sentry_sdk/ai/monitoring.py

@@ -94,7 +94,10 @@ async def async_wrapped(*args: "Any", **kwargs: "Any") -> "Any":
 
 
 def record_token_usage(
-    span: Span, prompt_tokens: "Optional[int]" = None, completion_tokens: "Optional[int]" = None, total_tokens: "Optional[int]" = None
+    span: Span,
+    prompt_tokens: "Optional[int]" = None,
+    completion_tokens: "Optional[int]" = None,
+    total_tokens: "Optional[int]" = None,
 ) -> None:
     ai_pipeline_name = get_ai_pipeline_name()
     if ai_pipeline_name:

sentry_sdk/client.py

@@ -192,14 +192,14 @@ def capture_session(self, *args: Any, **kwargs: Any) -> None:
     if TYPE_CHECKING:
 
         @overload
-        def get_integration(self, name_or_class: str) -> Optional["Integration"]:
-            ...
+        def get_integration(self, name_or_class: str) -> Optional["Integration"]: ...
 
         @overload
-        def get_integration(self, name_or_class: "type[I]") -> Optional["I"]:
-            ...
+        def get_integration(self, name_or_class: "type[I]") -> Optional["I"]: ...
 
-    def get_integration(self, name_or_class: Union[str, "type[Integration]"]) -> Optional["Integration"]:
+    def get_integration(
+        self, name_or_class: Union[str, "type[Integration]"]
+    ) -> Optional["Integration"]:
         return None
 
     def close(self, *args: Any, **kwargs: Any) -> None:
@@ -246,7 +246,9 @@ def __setstate__(self, state: Any) -> None:
         self.options = state["options"]
         self._init_impl()
 
-    def _setup_instrumentation(self, functions_to_trace: "Sequence[Dict[str, str]]") -> None:
+    def _setup_instrumentation(
+        self, functions_to_trace: "Sequence[Dict[str, str]]"
+    ) -> None:
         """
         Instruments the functions given in the list `functions_to_trace` with the `@sentry_sdk.tracing.trace` decorator.
         """
@@ -890,14 +892,14 @@ def capture_session(self, session: "Session") -> None:
     if TYPE_CHECKING:
 
         @overload
-        def get_integration(self, name_or_class: str) -> Optional["Integration"]:
-            ...
+        def get_integration(self, name_or_class: str) -> Optional["Integration"]: ...
 
         @overload
-        def get_integration(self, name_or_class: "type[I]") -> Optional["I"]:
-            ...
+        def get_integration(self, name_or_class: "type[I]") -> Optional["I"]: ...
 
-    def get_integration(self, name_or_class: Union[str, "Type[Integration]"]) -> Optional["Integration"]:
+    def get_integration(
+        self, name_or_class: Union[str, "Type[Integration]"]
+    ) -> Optional["Integration"]:
         """Returns the integration for this client by name or class.
         If the client does not have that integration then `None` is returned.
         """

sentry_sdk/integrations/__init__.py

@@ -34,7 +34,9 @@ def _generate_default_integrations_iterator(
     auto_enabling_integrations: "List[str]",
 ) -> "Callable[[bool], Iterator[Type[Integration]]]":
 
-    def iter_default_integrations(with_auto_enabling_integrations: bool) -> "Iterator[Type[Integration]]":
+    def iter_default_integrations(
+        with_auto_enabling_integrations: bool,
+    ) -> "Iterator[Type[Integration]]":
         """Returns an iterator of the default integration classes:"""
         from importlib import import_module
 
@@ -236,7 +238,11 @@ def setup_integrations(
     return integrations
 
 
-def _check_minimum_version(integration: "type[Integration]", version: "Optional[tuple[int, ...]]", package: "Optional[str]" = None) -> None:
+def _check_minimum_version(
+    integration: "type[Integration]",
+    version: "Optional[tuple[int, ...]]",
+    package: "Optional[str]" = None,
+) -> None:
     package = package or integration.identifier
 
     if version is None:

sentry_sdk/integrations/_asgi_common.py

@@ -31,7 +31,11 @@ def _get_headers(asgi_scope: "Any") -> "Dict[str, str]":
     return headers
 
 
-def _get_url(asgi_scope: "Dict[str, Any]", default_scheme: "Optional[Literal['ws', 'http']]" = None, host: "Optional[Union[AnnotatedValue, str]]" = None) -> str:
+def _get_url(
+    asgi_scope: "Dict[str, Any]",
+    default_scheme: "Optional[Literal['ws', 'http']]" = None,
+    host: "Optional[Union[AnnotatedValue, str]]" = None,
+) -> str:
     """
     Extract URL from the ASGI scope, without also including the querystring.
     """

sentry_sdk/integrations/_wsgi_common.py

@@ -199,7 +199,9 @@ def _is_json_content_type(ct: "Optional[str]") -> bool:
     )
 
 
-def _filter_headers(headers: "Mapping[str, str]") -> "Mapping[str, Union[AnnotatedValue, str]]":
+def _filter_headers(
+    headers: "Mapping[str, str]",
+) -> "Mapping[str, Union[AnnotatedValue, str]]":
     if should_send_default_pii():
         return headers
 

sentry_sdk/integrations/aiohttp.py

@@ -115,7 +115,9 @@ def setup_once() -> None:
 
         old_handle = Application._handle
 
-        async def sentry_app_handle(self: "Any", request: "Request", *args: "Any", **kwargs: "Any") -> "Any":
+        async def sentry_app_handle(
+            self: "Any", request: "Request", *args: "Any", **kwargs: "Any"
+        ) -> "Any":
             integration = sentry_sdk.get_client().get_integration(AioHttpIntegration)
             if integration is None:
                 return await old_handle(self, request, *args, **kwargs)
@@ -169,7 +171,9 @@ async def sentry_app_handle(self: "Any", request: "Request", *args: "Any", **kwa
         old_urldispatcher_resolve = UrlDispatcher.resolve
 
         @wraps(old_urldispatcher_resolve)
-        async def sentry_urldispatcher_resolve(self: "UrlDispatcher", request: "Request") -> "UrlMappingMatchInfo":
+        async def sentry_urldispatcher_resolve(
+            self: "UrlDispatcher", request: "Request"
+        ) -> "UrlMappingMatchInfo":
             rv = await old_urldispatcher_resolve(self, request)
 
             integration = sentry_sdk.get_client().get_integration(AioHttpIntegration)
@@ -214,7 +218,9 @@ def init(*args: "Any", **kwargs: "Any") -> None:
 
 def create_trace_config() -> "Any":
 
-    async def on_request_start(session: "Any", trace_config_ctx: "SimpleNamespace", params: "Any") -> None:
+    async def on_request_start(
+        session: "Any", trace_config_ctx: "SimpleNamespace", params: "Any"
+    ) -> None:
         if sentry_sdk.get_client().get_integration(AioHttpIntegration) is None:
             return
 
@@ -270,7 +276,9 @@ async def on_request_start(session: "Any", trace_config_ctx: "SimpleNamespace",
         trace_config_ctx.span = span
         trace_config_ctx.span_data = data
 
-    async def on_request_end(session: "Any", trace_config_ctx: "SimpleNamespace", params: "Any") -> None:
+    async def on_request_end(
+        session: "Any", trace_config_ctx: "SimpleNamespace", params: "Any"
+    ) -> None:
         if trace_config_ctx.span is None:
             return
 
@@ -299,7 +307,9 @@ async def on_request_end(session: "Any", trace_config_ctx: "SimpleNamespace", pa
     return trace_config
 
 
-def _make_request_processor(weak_request: "weakref.ReferenceType[Request]") -> "EventProcessor":
+def _make_request_processor(
+    weak_request: "weakref.ReferenceType[Request]",
+) -> "EventProcessor":
     def aiohttp_processor(
         event: "Event",
         hint: "dict[str, Tuple[type, BaseException, Any]]",
@@ -346,7 +356,9 @@ def _capture_exception() -> "ExcInfo":
 BODY_NOT_READ_MESSAGE = "[Can't show request body due to implementation details.]"
 
 
-def get_aiohttp_request_data(request: "Request") -> "Union[Optional[str], AnnotatedValue]":
+def get_aiohttp_request_data(
+    request: "Request",
+) -> "Union[Optional[str], AnnotatedValue]":
     bytes_body = request._read_bytes
 
     if bytes_body is not None:

sentry_sdk/integrations/anthropic.py

@@ -80,7 +80,12 @@ def _get_responses(content: "list[Any]") -> "list[dict[str, Any]]":
     return responses
 
 
-def _collect_ai_data(event: "MessageStreamEvent", input_tokens: int, output_tokens: int, content_blocks: "list[str]") -> "tuple[int, int, list[str]]":
+def _collect_ai_data(
+    event: "MessageStreamEvent",
+    input_tokens: int,
+    output_tokens: int,
+    content_blocks: "list[str]",
+) -> "tuple[int, int, list[str]]":
     """
     Count token usage and collect content blocks from the AI streaming response.
     """
@@ -106,7 +111,11 @@ def _collect_ai_data(event: "MessageStreamEvent", input_tokens: int, output_toke
 
 
 def _add_ai_data_to_span(
-    span: "Span", integration: "AnthropicIntegration", input_tokens: int, output_tokens: int, content_blocks: "list[str]"
+    span: "Span",
+    integration: "AnthropicIntegration",
+    input_tokens: int,
+    output_tokens: int,
+    content_blocks: "list[str]",
 ) -> None:
     """
     Add token usage and content blocks from the AI streaming response to the span.

sentry_sdk/integrations/ariadne.py

@@ -48,15 +48,21 @@ def _patch_graphql() -> None:
     old_handle_query_result = ariadne_graphql.handle_query_result
 
     @ensure_integration_enabled(AriadneIntegration, old_parse_query)
-    def _sentry_patched_parse_query(context_value: "Optional[Any]", query_parser: "Optional[QueryParser]", data: "Any") -> "DocumentNode":
+    def _sentry_patched_parse_query(
+        context_value: "Optional[Any]",
+        query_parser: "Optional[QueryParser]",
+        data: "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: "List[GraphQLError]", *args: "Any", **kwargs: "Any") -> "GraphQLResult":
+    def _sentry_patched_handle_graphql_errors(
+        errors: "List[GraphQLError]", *args: "Any", **kwargs: "Any"
+    ) -> "GraphQLResult":
         result = old_handle_errors(errors, *args, **kwargs)
 
         event_processor = _make_response_event_processor(result[1])
@@ -79,7 +85,9 @@ def _sentry_patched_handle_graphql_errors(errors: "List[GraphQLError]", *args: "
         return result
 
     @ensure_integration_enabled(AriadneIntegration, old_handle_query_result)
-    def _sentry_patched_handle_query_result(result: "Any", *args: "Any", **kwargs: "Any") -> "GraphQLResult":
+    def _sentry_patched_handle_query_result(
+        result: "Any", *args: "Any", **kwargs: "Any"
+    ) -> "GraphQLResult":
         query_result = old_handle_query_result(result, *args, **kwargs)
 
         event_processor = _make_response_event_processor(query_result[1])

sentry_sdk/integrations/arq.py

@@ -68,7 +68,9 @@ def patch_enqueue_job() -> None:
     old_enqueue_job = ArqRedis.enqueue_job
     original_kwdefaults = old_enqueue_job.__kwdefaults__
 
-    async def _sentry_enqueue_job(self: "ArqRedis", function: str, *args: "Any", **kwargs: "Any") -> "Optional[Job]":
+    async def _sentry_enqueue_job(
+        self: "ArqRedis", function: str, *args: "Any", **kwargs: "Any"
+    ) -> "Optional[Job]":
         integration = sentry_sdk.get_client().get_integration(ArqIntegration)
         if integration is None:
             return await old_enqueue_job(self, function, *args, **kwargs)
@@ -135,7 +137,9 @@ def _capture_exception(exc_info: "ExcInfo") -> None:
     sentry_sdk.capture_event(event, hint=hint)
 
 
-def _make_event_processor(ctx: "Dict[Any, Any]", *args: "Any", **kwargs: "Any") -> "EventProcessor":
+def _make_event_processor(
+    ctx: "Dict[Any, Any]", *args: "Any", **kwargs: "Any"
+) -> "EventProcessor":
     def event_processor(event: "Event", hint: "Hint") -> "Optional[Event]":
         with capture_internal_exceptions():
             scope = sentry_sdk.get_current_scope()
@@ -164,7 +168,9 @@ def event_processor(event: "Event", hint: "Hint") -> "Optional[Event]":
 
 
 def _wrap_coroutine(name: str, coroutine: "WorkerCoroutine") -> "WorkerCoroutine":
-    async def _sentry_coroutine(ctx: "Dict[Any, Any]", *args: "Any", **kwargs: "Any") -> "Any":
+    async def _sentry_coroutine(
+        ctx: "Dict[Any, Any]", *args: "Any", **kwargs: "Any"
+    ) -> "Any":
         integration = sentry_sdk.get_client().get_integration(ArqIntegration)
         if integration is None:
             return await coroutine(ctx, *args, **kwargs)

sentry_sdk/integrations/asgi.py

@@ -155,7 +155,9 @@ async def inner(receive: Any, send: Any) -> Any:
     async def _run_asgi3(self, scope: Any, receive: Any, send: Any) -> Any:
         return await self._run_app(scope, receive, send, asgi_version=3)
 
-    async def _run_original_app(self, scope: Any, receive: Any, send: Any, asgi_version: int) -> Any:
+    async def _run_original_app(
+        self, scope: Any, receive: Any, send: Any, asgi_version: int
+    ) -> Any:
         try:
             if asgi_version == 2:
                 return await self.app(scope)(receive, send)
@@ -166,7 +168,9 @@ async def _run_original_app(self, scope: Any, receive: Any, send: Any, asgi_vers
             _capture_exception(exc, mechanism_type=self.mechanism_type)
             raise exc from None
 
-    async def _run_app(self, scope: Any, receive: Any, send: Any, asgi_version: int) -> Any:
+    async def _run_app(
+        self, scope: Any, receive: Any, send: Any, asgi_version: int
+    ) -> Any:
         is_recursive_asgi_middleware = _asgi_middleware_applied.get(False)
         is_lifespan = scope["type"] == "lifespan"
         if is_recursive_asgi_middleware or is_lifespan:
@@ -220,7 +224,9 @@ async def _run_app(self, scope: Any, receive: Any, send: Any, asgi_version: int)
                                 logger.debug("[ASGI] Started transaction: %s", span)
                                 span.set_tag("asgi.type", ty)
 
-                            async def _sentry_wrapped_send(event: "Dict[str, Any]") -> Any:
+                            async def _sentry_wrapped_send(
+                                event: "Dict[str, Any]",
+                            ) -> Any:
                                 is_http_response = (
                                     event.get("type") == "http.response.start"
                                     and span is not None
@@ -237,7 +243,9 @@ async def _sentry_wrapped_send(event: "Dict[str, Any]") -> Any:
         finally:
             _asgi_middleware_applied.set(False)
 
-    def event_processor(self, event: "Event", hint: "Hint", asgi_scope: Any) -> "Optional[Event]":
+    def event_processor(
+        self, event: "Event", hint: "Hint", asgi_scope: Any
+    ) -> "Optional[Event]":
         request_data = event.get("request", {})
         request_data.update(_get_request_data(asgi_scope))
         event["request"] = deepcopy(request_data)
@@ -276,7 +284,9 @@ def event_processor(self, event: "Event", hint: "Hint", asgi_scope: Any) -> "Opt
     # data to your liking it's recommended to use the `before_send` callback
     # for that.
 
-    def _get_transaction_name_and_source(self, transaction_style: str, asgi_scope: Any) -> "Tuple[str, str]":
+    def _get_transaction_name_and_source(
+        self, transaction_style: str, asgi_scope: Any
+    ) -> "Tuple[str, str]":
         name = None
         source = SOURCE_FOR_STYLE[transaction_style]
         ty = asgi_scope.get("type")