Migrate typing for integrations - part 1 (#4529)

Author: sl0thentr0py

sentry_sdk/integrations/__init__.py

@@ -1,3 +1,4 @@
+from __future__ import annotations
 from abc import ABC, abstractmethod
 from threading import Lock
 
@@ -23,20 +24,20 @@
 _installer_lock = Lock()
 
 # Set of all integration identifiers we have attempted to install
-_processed_integrations = set()  # type: Set[str]
+_processed_integrations: Set[str] = set()
 
 # Set of all integration identifiers we have actually installed
-_installed_integrations = set()  # type: Set[str]
+_installed_integrations: Set[str] = set()
 
 
 def _generate_default_integrations_iterator(
-    integrations,  # type: List[str]
-    auto_enabling_integrations,  # type: List[str]
-):
-    # type: (...) -> Callable[[bool], Iterator[Type[Integration]]]
+    integrations: List[str],
+    auto_enabling_integrations: List[str],
+) -> Callable[[bool], Iterator[Type[Integration]]]:
 
-    def iter_default_integrations(with_auto_enabling_integrations):
-        # type: (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
 
@@ -165,12 +166,13 @@ def iter_default_integrations(with_auto_enabling_integrations):
 
 
 def setup_integrations(
-    integrations,
-    with_defaults=True,
-    with_auto_enabling_integrations=False,
-    disabled_integrations=None,
-):
-    # type: (Sequence[Integration], bool, bool, Optional[Sequence[Union[type[Integration], Integration]]]) -> Dict[str, Integration]
+    integrations: Sequence[Integration],
+    with_defaults: bool = True,
+    with_auto_enabling_integrations: bool = False,
+    disabled_integrations: Optional[
+        Sequence[Union[type[Integration], Integration]]
+    ] = None,
+) -> Dict[str, Integration]:
     """
     Given a list of integration instances, this installs them all.
 
@@ -239,8 +241,11 @@ def setup_integrations(
     return integrations
 
 
-def _check_minimum_version(integration, version, package=None):
-    # type: (type[Integration], Optional[tuple[int, ...]], Optional[str]) -> 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:
@@ -276,13 +281,12 @@ class Integration(ABC):
     install = None
     """Legacy method, do not implement."""
 
-    identifier = None  # type: str
+    identifier: str
     """String unique ID of integration type"""
 
     @staticmethod
     @abstractmethod
-    def setup_once():
-        # type: () -> None
+    def setup_once() -> None:
         """
         Initialize the integration.
 

sentry_sdk/integrations/_asgi_common.py

@@ -1,26 +1,25 @@
+from __future__ import annotations
 import urllib
 
 from sentry_sdk.scope import should_send_default_pii
 from sentry_sdk.integrations._wsgi_common import _filter_headers
 
-from typing import TYPE_CHECKING, cast
+from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
     from typing import Any
     from typing import Dict
     from typing import Optional
     from typing import Union
-    from typing_extensions import Literal
 
     from sentry_sdk.utils import AnnotatedValue
 
 
-def _get_headers(asgi_scope):
-    # type: (Any) -> Dict[str, str]
+def _get_headers(asgi_scope: Any) -> Dict[str, str]:
     """
     Extract headers from the ASGI scope, in the format that the Sentry protocol expects.
     """
-    headers = {}  # type: Dict[str, str]
+    headers: Dict[str, str] = {}
     for raw_key, raw_value in asgi_scope.get("headers", {}):
         key = raw_key.decode("latin-1")
         value = raw_value.decode("latin-1")
@@ -32,12 +31,16 @@ def _get_headers(asgi_scope):
     return headers
 
 
-def _get_url(asgi_scope, default_scheme=None, host=None):
-    # type: (Dict[str, Any], Optional[Literal["ws", "http"]], Optional[Union[AnnotatedValue, str]]) -> str
+def _get_url(
+    asgi_scope: Dict[str, Any],
+    host: Optional[Union[AnnotatedValue, str]] = None,
+) -> str:
     """
     Extract URL from the ASGI scope, without also including the querystring.
     """
-    scheme = cast(str, asgi_scope.get("scheme", default_scheme))
+    scheme = asgi_scope.get(
+        "scheme", "http" if asgi_scope.get("type") == "http" else "ws"
+    )
     server = asgi_scope.get("server", None)
     path = asgi_scope.get("root_path", "") + asgi_scope.get("path", "")
 
@@ -53,8 +56,7 @@ def _get_url(asgi_scope, default_scheme=None, host=None):
     return path
 
 
-def _get_query(asgi_scope):
-    # type: (Any) -> Any
+def _get_query(asgi_scope: Any) -> Any:
     """
     Extract querystring from the ASGI scope, in the format that the Sentry protocol expects.
     """
@@ -64,8 +66,7 @@ def _get_query(asgi_scope):
     return urllib.parse.unquote(qs.decode("latin-1"))
 
 
-def _get_ip(asgi_scope):
-    # type: (Any) -> str
+def _get_ip(asgi_scope: Any) -> str:
     """
     Extract IP Address from the ASGI scope based on request headers with fallback to scope client.
     """
@@ -83,22 +84,19 @@ def _get_ip(asgi_scope):
     return asgi_scope.get("client")[0]
 
 
-def _get_request_data(asgi_scope):
-    # type: (Any) -> Dict[str, Any]
+def _get_request_data(asgi_scope: Any) -> Dict[str, Any]:
     """
     Returns data related to the HTTP request from the ASGI scope.
     """
-    request_data = {}  # type: Dict[str, Any]
+    request_data: Dict[str, Any] = {}
     ty = asgi_scope["type"]
     if ty in ("http", "websocket"):
         request_data["method"] = asgi_scope.get("method")
 
         request_data["headers"] = headers = _filter_headers(_get_headers(asgi_scope))
         request_data["query_string"] = _get_query(asgi_scope)
 
-        request_data["url"] = _get_url(
-            asgi_scope, "http" if ty == "http" else "ws", headers.get("host")
-        )
+        request_data["url"] = _get_url(asgi_scope, headers.get("host"))
 
     client = asgi_scope.get("client")
     if client and should_send_default_pii():

sentry_sdk/integrations/_wsgi_common.py

@@ -1,3 +1,4 @@
+from __future__ import annotations
 import json
 from copy import deepcopy
 
@@ -50,8 +51,9 @@
 )
 
 
-def request_body_within_bounds(client, content_length):
-    # type: (Optional[sentry_sdk.client.BaseClient], int) -> bool
+def request_body_within_bounds(
+    client: Optional[sentry_sdk.client.BaseClient], content_length: int
+) -> bool:
     if client is None:
         return False
 
@@ -73,17 +75,15 @@ class RequestExtractor:
     # it. Only some child classes implement all methods that raise
     # NotImplementedError in this class.
 
-    def __init__(self, request):
-        # type: (Any) -> None
+    def __init__(self, request: Any) -> None:
         self.request = request
 
-    def extract_into_event(self, event):
-        # type: (Event) -> None
+    def extract_into_event(self, event: Event) -> None:
         client = sentry_sdk.get_client()
         if not client.is_active():
             return
 
-        data = None  # type: Optional[Union[AnnotatedValue, Dict[str, Any]]]
+        data: Optional[Union[AnnotatedValue, Dict[str, Any]]] = None
 
         content_length = self.content_length()
         request_info = event.get("request", {})
@@ -119,27 +119,22 @@ def extract_into_event(self, event):
 
         event["request"] = deepcopy(request_info)
 
-    def content_length(self):
-        # type: () -> int
+    def content_length(self) -> int:
         try:
             return int(self.env().get("CONTENT_LENGTH", 0))
         except ValueError:
             return 0
 
-    def cookies(self):
-        # type: () -> MutableMapping[str, Any]
+    def cookies(self) -> MutableMapping[str, Any]:
         raise NotImplementedError()
 
-    def raw_data(self):
-        # type: () -> Optional[Union[str, bytes]]
+    def raw_data(self) -> Optional[Union[str, bytes]]:
         raise NotImplementedError()
 
-    def form(self):
-        # type: () -> Optional[Dict[str, Any]]
+    def form(self) -> Optional[Dict[str, Any]]:
         raise NotImplementedError()
 
-    def parsed_body(self):
-        # type: () -> Optional[Dict[str, Any]]
+    def parsed_body(self) -> Optional[Dict[str, Any]]:
         try:
             form = self.form()
         except Exception:
@@ -161,12 +156,10 @@ def parsed_body(self):
 
         return self.json()
 
-    def is_json(self):
-        # type: () -> bool
+    def is_json(self) -> bool:
         return _is_json_content_type(self.env().get("CONTENT_TYPE"))
 
-    def json(self):
-        # type: () -> Optional[Any]
+    def json(self) -> Optional[Any]:
         try:
             if not self.is_json():
                 return None
@@ -190,21 +183,17 @@ def json(self):
 
         return None
 
-    def files(self):
-        # type: () -> Optional[Dict[str, Any]]
+    def files(self) -> Optional[Dict[str, Any]]:
         raise NotImplementedError()
 
-    def size_of_file(self, file):
-        # type: (Any) -> int
+    def size_of_file(self, file: Any) -> int:
         raise NotImplementedError()
 
-    def env(self):
-        # type: () -> Dict[str, Any]
+    def env(self) -> Dict[str, Any]:
         raise NotImplementedError()
 
 
-def _is_json_content_type(ct):
-    # type: (Optional[str]) -> bool
+def _is_json_content_type(ct: Optional[str]) -> bool:
     mt = (ct or "").split(";", 1)[0]
     return (
         mt == "application/json"
@@ -213,8 +202,9 @@ def _is_json_content_type(ct):
     )
 
 
-def _filter_headers(headers):
-    # type: (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
 
@@ -228,8 +218,7 @@ def _filter_headers(headers):
     }
 
 
-def _request_headers_to_span_attributes(headers):
-    # type: (dict[str, str]) -> dict[str, str]
+def _request_headers_to_span_attributes(headers: dict[str, str]) -> dict[str, str]:
     attributes = {}
 
     headers = _filter_headers(headers)