Migrate typing for core files

Author: sl0thentr0py

sentry_sdk/_compat.py

@@ -1,31 +1,22 @@
+from __future__ import annotations
 import sys
-
-from typing import TYPE_CHECKING
-
-if TYPE_CHECKING:
-    from typing import Any
-    from typing import TypeVar
-
-    T = TypeVar("T")
+from typing import Any
 
 
 PY38 = sys.version_info[0] == 3 and sys.version_info[1] >= 8
 PY310 = sys.version_info[0] == 3 and sys.version_info[1] >= 10
 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
@@ -45,8 +36,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

@@ -1,23 +1,18 @@
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Optional, Any
 
 import sentry_sdk
 from sentry_sdk.consts import ClientConstructor
 from sentry_sdk.opentelemetry.scope import setup_scope_context_management
 
-if TYPE_CHECKING:
-    from typing import Any, Optional
-
 
-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) -> None
+def _init(*args: Optional[str], **kwargs: Any) -> None:
     """Initializes the SDK and optionally integrations.
 
     This takes the same arguments as the client constructor.

sentry_sdk/_log_batcher.py

@@ -1,3 +1,4 @@
+from __future__ import annotations
 import os
 import random
 import threading
@@ -15,23 +16,18 @@ class LogBatcher:
     MAX_LOGS_BEFORE_FLUSH = 100
     FLUSH_WAIT_TIME = 5.0
 
-    def __init__(
-        self,
-        capture_func,  # type: Callable[[Envelope], None]
-    ):
-        # type: (...) -> None
-        self._log_buffer = []  # type: List[Log]
+    def __init__(self, capture_func: Callable[[Envelope], None]) -> None:
+        self._log_buffer: List[Log] = []
         self._capture_func = capture_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.
         """
@@ -63,18 +59,13 @@ 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
+    def add(self, log: Log) -> None:
         if not self._ensure_thread() or self._flusher is None:
             return None
 
@@ -83,24 +74,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):
@@ -128,8 +115,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

@@ -1,24 +1,19 @@
-from typing import TYPE_CHECKING
-
-if TYPE_CHECKING:
-    from typing import Any
+from typing import Any
 
 
 _SENTINEL = object()
 
 
 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 +24,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 +36,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/_queue.py

@@ -76,10 +76,8 @@
 from collections import deque
 from time import time
 
-from typing import TYPE_CHECKING
+from typing import Any
 
-if TYPE_CHECKING:
-    from typing import Any
 
 __all__ = ["EmptyError", "FullError", "Queue"]
 
@@ -275,7 +273,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/_werkzeug.py

@@ -32,12 +32,7 @@
 SUCH DAMAGE.
 """
 
-from typing import TYPE_CHECKING
-
-if TYPE_CHECKING:
-    from typing import Dict
-    from typing import Iterator
-    from typing import Tuple
+from typing import Dict, Iterator, Tuple
 
 
 #
@@ -47,8 +42,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 +61,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.
     """