Convert sentry_sdk type annotations into the modern format

Author: zsol

sentry_sdk/_compat.py

@@ -15,18 +15,15 @@
 PY311 = sys.version_info[0] == 3 and sys.version_info[1] >= 11
 
 
-def with_metaclass(meta, *bases):
-    # type: (Any, *Any) -> Any
+def with_metaclass(meta: "Any", *bases: "Any") -> "Any":
     class MetaClass(type):
-        def __new__(metacls, name, this_bases, d):
-            # type: (Any, Any, Any, Any) -> Any
+        def __new__(metacls: "Any", name: "Any", this_bases: "Any", d: "Any") -> "Any":
             return meta(name, bases, d)
 
     return type.__new__(MetaClass, "temporary_class", (), {})
 
 
-def check_uwsgi_thread_support():
-    # type: () -> bool
+def check_uwsgi_thread_support() -> bool:
     # We check two things here:
     #
     # 1. uWSGI doesn't run in threaded mode by default -- issue a warning if
@@ -46,8 +43,7 @@ def check_uwsgi_thread_support():
 
     from sentry_sdk.consts import FALSE_VALUES
 
-    def enabled(option):
-        # type: (str) -> bool
+    def enabled(option: str) -> bool:
         value = opt.get(option, False)
         if isinstance(value, bool):
             return value

sentry_sdk/_init_implementation.py

@@ -18,12 +18,10 @@ class _InitGuard:
         "functionality, and we will remove it in the next major release."
     )
 
-    def __init__(self, client):
-        # type: (sentry_sdk.Client) -> None
+    def __init__(self, client: "sentry_sdk.Client") -> None:
         self._client = client
 
-    def __enter__(self):
-        # type: () -> _InitGuard
+    def __enter__(self) -> "_InitGuard":
         warnings.warn(
             self._CONTEXT_MANAGER_DEPRECATION_WARNING_MESSAGE,
             stacklevel=2,
@@ -32,8 +30,7 @@ def __enter__(self):
 
         return self
 
-    def __exit__(self, exc_type, exc_value, tb):
-        # type: (Any, Any, Any) -> None
+    def __exit__(self, exc_type: "Any", exc_value: "Any", tb: "Any") -> None:
         warnings.warn(
             self._CONTEXT_MANAGER_DEPRECATION_WARNING_MESSAGE,
             stacklevel=2,
@@ -45,16 +42,14 @@ def __exit__(self, exc_type, exc_value, tb):
             c.close()
 
 
-def _check_python_deprecations():
-    # type: () -> None
+def _check_python_deprecations() -> None:
     # Since we're likely to deprecate Python versions in the future, I'm keeping
     # this handy function around. Use this to detect the Python version used and
     # to output logger.warning()s if it's deprecated.
     pass
 
 
-def _init(*args, **kwargs):
-    # type: (*Optional[str], **Any) -> ContextManager[Any]
+def _init(*args: "Optional[str]", **kwargs: "Any") -> "ContextManager[Any]":
     """Initializes the SDK and optionally integrations.
 
     This takes the same arguments as the client constructor.

sentry_sdk/_log_batcher.py

@@ -18,23 +18,21 @@ class LogBatcher:
 
     def __init__(
         self,
-        capture_func,  # type: Callable[[Envelope], None]
-        record_lost_func,  # type: Callable[..., None]
-    ):
-        # type: (...) -> None
-        self._log_buffer = []  # type: List[Log]
+        capture_func: "Callable[[Envelope], None]",
+        record_lost_func: "Callable[..., None]",
+    ) -> None:
+        self._log_buffer: List[Log] = []
         self._capture_func = capture_func
         self._record_lost_func = record_lost_func
         self._running = True
         self._lock = threading.Lock()
 
-        self._flush_event = threading.Event()  # type: threading.Event
+        self._flush_event: "threading.Event" = threading.Event()
 
-        self._flusher = None  # type: Optional[threading.Thread]
-        self._flusher_pid = None  # type: Optional[int]
+        self._flusher: "Optional[threading.Thread]" = None
+        self._flusher_pid: "Optional[int]" = None
 
-    def _ensure_thread(self):
-        # type: (...) -> bool
+    def _ensure_thread(self) -> bool:
         """For forking processes we might need to restart this thread.
         This ensures that our process actually has that thread running.
         """
@@ -66,18 +64,16 @@ def _ensure_thread(self):
 
         return True
 
-    def _flush_loop(self):
-        # type: (...) -> None
+    def _flush_loop(self) -> None:
         while self._running:
             self._flush_event.wait(self.FLUSH_WAIT_TIME + random.random())
             self._flush_event.clear()
             self._flush()
 
     def add(
         self,
-        log,  # type: Log
-    ):
-        # type: (...) -> None
+        log: "Log",
+    ) -> None:
         if not self._ensure_thread() or self._flusher is None:
             return None
 
@@ -106,24 +102,20 @@ def add(
             if len(self._log_buffer) >= self.MAX_LOGS_BEFORE_FLUSH:
                 self._flush_event.set()
 
-    def kill(self):
-        # type: (...) -> None
+    def kill(self) -> None:
         if self._flusher is None:
             return
 
         self._running = False
         self._flush_event.set()
         self._flusher = None
 
-    def flush(self):
-        # type: (...) -> None
+    def flush(self) -> None:
         self._flush()
 
     @staticmethod
-    def _log_to_transport_format(log):
-        # type: (Log) -> Any
-        def format_attribute(val):
-            # type: (int | float | str | bool) -> Any
+    def _log_to_transport_format(log: "Log") -> "Any":
+        def format_attribute(val: int | float | str | bool) -> Any:
             if isinstance(val, bool):
                 return {"value": val, "type": "boolean"}
             if isinstance(val, int):
@@ -151,9 +143,7 @@ def format_attribute(val):
 
         return res
 
-    def _flush(self):
-        # type: (...) -> Optional[Envelope]
-
+    def _flush(self) -> "Optional[Envelope]":
         envelope = Envelope(
             headers={"sent_at": format_timestamp(datetime.now(timezone.utc))}
         )

sentry_sdk/_lru_cache.py

@@ -8,17 +8,15 @@
 
 
 class LRUCache:
-    def __init__(self, max_size):
-        # type: (int) -> None
+    def __init__(self, max_size: int) -> None:
         if max_size <= 0:
             raise AssertionError(f"invalid max_size: {max_size}")
         self.max_size = max_size
-        self._data = {}  # type: dict[Any, Any]
+        self._data: "dict[Any, Any]" = {}
         self.hits = self.misses = 0
         self.full = False
 
-    def set(self, key, value):
-        # type: (Any, Any) -> None
+    def set(self, key: "Any", value: "Any") -> None:
         current = self._data.pop(key, _SENTINEL)
         if current is not _SENTINEL:
             self._data[key] = value
@@ -29,8 +27,7 @@ def set(self, key, value):
             self._data[key] = value
         self.full = len(self._data) >= self.max_size
 
-    def get(self, key, default=None):
-        # type: (Any, Any) -> Any
+    def get(self, key: "Any", default: "Any" = None) -> "Any":
         try:
             ret = self._data.pop(key)
         except KeyError:
@@ -42,6 +39,5 @@ def get(self, key, default=None):
 
         return ret
 
-    def get_all(self):
-        # type: () -> list[tuple[Any, Any]]
+    def get_all(self) -> "list[tuple[Any, Any]]":
         return list(self._data.items())

sentry_sdk/_metrics_batcher.py

@@ -18,23 +18,21 @@ class MetricsBatcher:
 
     def __init__(
         self,
-        capture_func,  # type: Callable[[Envelope], None]
-        record_lost_func,  # type: Callable[..., None]
-    ):
-        # type: (...) -> None
-        self._metric_buffer = []  # type: List[Metric]
+        capture_func: "Callable[[Envelope], None]",
+        record_lost_func: "Callable[..., None]",
+    ) -> None:
+        self._metric_buffer: List[Metric] = []
         self._capture_func = capture_func
         self._record_lost_func = record_lost_func
         self._running = True
         self._lock = threading.Lock()
 
-        self._flush_event = threading.Event()  # type: threading.Event
+        self._flush_event: "threading.Event" = threading.Event()
 
-        self._flusher = None  # type: Optional[threading.Thread]
-        self._flusher_pid = None  # type: Optional[int]
+        self._flusher: "Optional[threading.Thread]" = None
+        self._flusher_pid: "Optional[int]" = None
 
-    def _ensure_thread(self):
-        # type: (...) -> bool
+    def _ensure_thread(self) -> bool:
         if not self._running:
             return False
 
@@ -59,18 +57,16 @@ def _ensure_thread(self):
 
         return True
 
-    def _flush_loop(self):
-        # type: (...) -> None
+    def _flush_loop(self) -> None:
         while self._running:
             self._flush_event.wait(self.FLUSH_WAIT_TIME + random.random())
             self._flush_event.clear()
             self._flush()
 
     def add(
         self,
-        metric,  # type: Metric
-    ):
-        # type: (...) -> None
+        metric: "Metric",
+    ) -> None:
         if not self._ensure_thread() or self._flusher is None:
             return None
 
@@ -87,24 +83,20 @@ def add(
             if len(self._metric_buffer) >= self.MAX_METRICS_BEFORE_FLUSH:
                 self._flush_event.set()
 
-    def kill(self):
-        # type: (...) -> None
+    def kill(self) -> None:
         if self._flusher is None:
             return
 
         self._running = False
         self._flush_event.set()
         self._flusher = None
 
-    def flush(self):
-        # type: (...) -> None
+    def flush(self) -> None:
         self._flush()
 
     @staticmethod
-    def _metric_to_transport_format(metric):
-        # type: (Metric) -> Any
-        def format_attribute(val):
-            # type: (Union[int, float, str, bool]) -> Any
+    def _metric_to_transport_format(metric: "Metric") -> "Any":
+        def format_attribute(val: Union[int, float, str, bool]) -> Any:
             if isinstance(val, bool):
                 return {"value": val, "type": "boolean"}
             if isinstance(val, int):
@@ -134,9 +126,7 @@ def format_attribute(val):
 
         return res
 
-    def _flush(self):
-        # type: (...) -> Optional[Envelope]
-
+    def _flush(self) -> "Optional[Envelope]":
         envelope = Envelope(
             headers={"sent_at": format_timestamp(datetime.now(timezone.utc))}
         )

sentry_sdk/_queue.py

@@ -275,7 +275,7 @@ def get_nowait(self):
 
     # Initialize the queue representation
     def _init(self, maxsize):
-        self.queue = deque()  # type: Any
+        self.queue: "Any" = deque()
 
     def _qsize(self):
         return len(self.queue)

sentry_sdk/_types.py

@@ -18,32 +18,27 @@ class AnnotatedValue:
 
     __slots__ = ("value", "metadata")
 
-    def __init__(self, value, metadata):
-        # type: (Optional[Any], Dict[str, Any]) -> None
+    def __init__(self, value: "Optional[Any]", metadata: "Dict[str, Any]") -> None:
         self.value = value
         self.metadata = metadata
 
-    def __eq__(self, other):
-        # type: (Any) -> bool
+    def __eq__(self, other: "Any") -> bool:
         if not isinstance(other, AnnotatedValue):
             return False
 
         return self.value == other.value and self.metadata == other.metadata
 
-    def __str__(self):
-        # type: (AnnotatedValue) -> str
+    def __str__(self: "AnnotatedValue") -> str:
         return str({"value": str(self.value), "metadata": str(self.metadata)})
 
-    def __len__(self):
-        # type: (AnnotatedValue) -> int
+    def __len__(self: "AnnotatedValue") -> int:
         if self.value is not None:
             return len(self.value)
         else:
             return 0
 
     @classmethod
-    def removed_because_raw_data(cls):
-        # type: () -> AnnotatedValue
+    def removed_because_raw_data(cls) -> "AnnotatedValue":
         """The value was removed because it could not be parsed. This is done for request body values that are not json nor a form."""
         return AnnotatedValue(
             value="",
@@ -58,8 +53,7 @@ def removed_because_raw_data(cls):
         )
 
     @classmethod
-    def removed_because_over_size_limit(cls, value=""):
-        # type: (Any) -> AnnotatedValue
+    def removed_because_over_size_limit(cls, value: "Any" = "") -> "AnnotatedValue":
         """
         The actual value was removed because the size of the field exceeded the configured maximum size,
         for example specified with the max_request_body_size sdk option.
@@ -77,8 +71,7 @@ def removed_because_over_size_limit(cls, value=""):
         )
 
     @classmethod
-    def substituted_because_contains_sensitive_data(cls):
-        # type: () -> AnnotatedValue
+    def substituted_because_contains_sensitive_data(cls) -> "AnnotatedValue":
         """The actual value was removed because it contained sensitive information."""
         return AnnotatedValue(
             value=SENSITIVE_DATA_SUBSTITUTE,

sentry_sdk/_werkzeug.py

@@ -47,8 +47,7 @@
 # We need this function because Django does not give us a "pure" http header
 # dict. So we might as well use it for all WSGI integrations.
 #
-def _get_headers(environ):
-    # type: (Dict[str, str]) -> Iterator[Tuple[str, str]]
+def _get_headers(environ: "Dict[str, str]") -> "Iterator[Tuple[str, str]]":
     """
     Returns only proper HTTP headers.
     """
@@ -67,8 +66,7 @@ def _get_headers(environ):
 # `get_host` comes from `werkzeug.wsgi.get_host`
 # https://github.com/pallets/werkzeug/blob/1.0.1/src/werkzeug/wsgi.py#L145
 #
-def get_host(environ, use_x_forwarded_for=False):
-    # type: (Dict[str, str], bool) -> str
+def get_host(environ: "Dict[str, str]", use_x_forwarded_for: bool = False) -> str:
     """
     Return the host for the given WSGI environment.
     """