Run ast-grep rule

Author: zsol

sentry_sdk/_log_batcher.py

@@ -21,7 +21,7 @@ def __init__(
         capture_func: "Callable[[Envelope], None]",
         record_lost_func: "Callable[..., None]",
     ) -> None:
-        self._log_buffer: List[Log] = []
+        self._log_buffer: "List[Log]" = []
         self._capture_func = capture_func
         self._record_lost_func = record_lost_func
         self._running = True
@@ -115,7 +115,7 @@ def flush(self) -> None:
 
     @staticmethod
     def _log_to_transport_format(log: "Log") -> "Any":
-        def format_attribute(val: int | float | str | bool) -> Any:
+        def format_attribute(val: int | float | str | bool) -> "Any":
             if isinstance(val, bool):
                 return {"value": val, "type": "boolean"}
             if isinstance(val, int):

sentry_sdk/_metrics_batcher.py

@@ -21,7 +21,7 @@ def __init__(
         capture_func: "Callable[[Envelope], None]",
         record_lost_func: "Callable[..., None]",
     ) -> None:
-        self._metric_buffer: List[Metric] = []
+        self._metric_buffer: "List[Metric]" = []
         self._capture_func = capture_func
         self._record_lost_func = record_lost_func
         self._running = True
@@ -96,7 +96,7 @@ def flush(self) -> None:
 
     @staticmethod
     def _metric_to_transport_format(metric: "Metric") -> "Any":
-        def format_attribute(val: Union[int, float, str, bool]) -> Any:
+        def format_attribute(val: "Union[int, float, str, bool]") -> "Any":
             if isinstance(val, bool):
                 return {"value": val, "type": "boolean"}
             if isinstance(val, int):

sentry_sdk/_types.py

@@ -109,7 +109,7 @@ def substituted_because_contains_sensitive_data(cls) -> "AnnotatedValue":
     class SDKInfo(TypedDict):
         name: str
         version: str
-        packages: Sequence[Mapping[str, str]]
+        packages: "Sequence[Mapping[str, str]]"
 
     # "critical" is an alias of "fatal" recognized by Relay
     LogLevelStr = Literal["fatal", "critical", "error", "warning", "info", "debug"]

sentry_sdk/ai/monitoring.py

@@ -26,8 +26,8 @@ def get_ai_pipeline_name() -> "Optional[str]":
 
 
 def ai_track(description: str, **span_kwargs: "Any") -> "Callable[[F], F]":
-    def decorator(f: F) -> F:
-        def sync_wrapped(*args: Any, **kwargs: Any) -> Any:
+    def decorator(f: "F") -> "F":
+        def sync_wrapped(*args: "Any", **kwargs: "Any") -> "Any":
             curr_pipeline = _ai_pipeline_name.get()
             op = span_kwargs.pop("op", "ai.run" if curr_pipeline else "ai.pipeline")
 

sentry_sdk/client.py

@@ -180,7 +180,7 @@ class BaseClient:
     spotlight: "Optional[SpotlightClient]" = None
 
     def __init__(self, options: "Optional[Dict[str, Any]]" = None) -> None:
-        self.options: Dict[str, Any] = (
+        self.options: "Dict[str, Any]" = (
             options if options is not None else DEFAULT_OPTIONS
         )
 
@@ -495,7 +495,7 @@ def _prepare_event(
         hint: "Hint",
         scope: "Optional[Scope]",
     ) -> "Optional[Event]":
-        previous_total_spans: Optional[int] = None
+        previous_total_spans: "Optional[int]" = None
         previous_total_breadcrumbs: "Optional[int]" = None
 
         if event.get("timestamp") is None:

sentry_sdk/crons/decorator.py

@@ -122,15 +122,15 @@ def _async_wrapper(
         self, fn: "Callable[P, Awaitable[Any]]"
     ) -> "Callable[P, Awaitable[Any]]":
         @wraps(fn)
-        async def inner(*args: "P.args", **kwargs: "P.kwargs") -> R:
+        async def inner(*args: "P.args", **kwargs: "P.kwargs") -> "R":
             with self:
                 return await fn(*args, **kwargs)
 
         return inner
 
     def _sync_wrapper(self, fn: "Callable[P, R]") -> "Callable[P, R]":
         @wraps(fn)
-        def inner(*args: "P.args", **kwargs: "P.kwargs") -> R:
+        def inner(*args: "P.args", **kwargs: "P.kwargs") -> "R":
             with self:
                 return fn(*args, **kwargs)
 

sentry_sdk/integrations/__init__.py

@@ -27,7 +27,7 @@ def _generate_default_integrations_iterator(
 ) -> "Callable[[bool], Iterator[Type[Integration]]]":
     def iter_default_integrations(
         with_auto_enabling_integrations: bool,
-    ) -> Iterator[Type[Integration]]:
+    ) -> "Iterator[Type[Integration]]":
         """Returns an iterator of the default integration classes:"""
         from importlib import import_module
 

sentry_sdk/integrations/aiohttp.py

@@ -219,9 +219,9 @@ def init(*args: "Any", **kwargs: "Any") -> None:
 
 def create_trace_config() -> "TraceConfig":
     async def on_request_start(
-        session: ClientSession,
-        trace_config_ctx: SimpleNamespace,
-        params: TraceRequestStartParams,
+        session: "ClientSession",
+        trace_config_ctx: "SimpleNamespace",
+        params: "TraceRequestStartParams",
     ) -> None:
         if sentry_sdk.get_client().get_integration(AioHttpIntegration) is None:
             return
@@ -296,9 +296,9 @@ def _make_request_processor(
     weak_request: "weakref.ReferenceType[Request]",
 ) -> "EventProcessor":
     def aiohttp_processor(
-        event: Event,
-        hint: dict[str, Tuple[type, BaseException, Any]],
-    ) -> Event:
+        event: "Event",
+        hint: "dict[str, Tuple[type, BaseException, Any]]",
+    ) -> "Event":
         request = weak_request()
         if request is None:
             return event

sentry_sdk/integrations/anthropic.py

@@ -374,7 +374,7 @@ async def new_iterator_async() -> "AsyncIterator[MessageStreamEvent]":
 
 
 def _wrap_message_create(f: "Any") -> "Any":
-    def _execute_sync(f: Any, *args: Any, **kwargs: Any) -> Any:
+    def _execute_sync(f: "Any", *args: "Any", **kwargs: "Any") -> "Any":
         gen = _sentry_patched_create_common(f, *args, **kwargs)
 
         try:
@@ -410,7 +410,7 @@ def _sentry_patched_create_sync(*args: "Any", **kwargs: "Any") -> "Any":
 
 
 def _wrap_message_create_async(f: "Any") -> "Any":
-    async def _execute_async(f: Any, *args: Any, **kwargs: Any) -> Any:
+    async def _execute_async(f: "Any", *args: "Any", **kwargs: "Any") -> "Any":
         gen = _sentry_patched_create_common(f, *args, **kwargs)
 
         try:

sentry_sdk/integrations/argv.py

@@ -18,7 +18,7 @@ class ArgvIntegration(Integration):
     @staticmethod
     def setup_once() -> None:
         @add_global_event_processor
-        def processor(event: Event, hint: Optional[Hint]) -> Optional[Event]:
+        def processor(event: "Event", hint: "Optional[Hint]") -> "Optional[Event]":
             if sentry_sdk.get_client().get_integration(ArgvIntegration) is not None:
                 extra = event.setdefault("extra", {})
                 # If some event processor decided to set extra to e.g. an