fun with python types

Author: BYK

sentry_sdk/transport.py

@@ -35,6 +35,7 @@
     from typing import List
     from typing import Mapping
     from typing import Optional
+    from typing import Self
     from typing import Tuple
     from typing import Type
     from typing import Union
@@ -67,20 +68,16 @@ class Transport(ABC):
 
     parsed_dsn = None  # type: Optional[Dsn]
 
-    def __init__(
-        self, options=None  # type: Optional[Dict[str, Any]]
-    ):
-        # type: (...) -> None
+    def __init__(self, options=None):
+        # type: (Self, Optional[Dict[str, Any]]) -> None
         self.options = options
         if options and options["dsn"] is not None and options["dsn"]:
             self.parsed_dsn = Dsn(options["dsn"])
         else:
             self.parsed_dsn = None
 
-    def capture_event(
-        self, event  # type: Event
-    ):
-        # type: (...) -> None
+    def capture_event(self, event):
+        # type: (Self, Event) -> None
         """
         DEPRECATED: Please use capture_envelope instead.
 
@@ -99,25 +96,23 @@ def capture_event(
         self.capture_envelope(envelope)
 
     @abstractmethod
-    def capture_envelope(
-        self, envelope  # type: Envelope
-    ):
-        # type: (...) -> None
+    def capture_envelope(self, envelope):
+        # type: (Self, Envelope) -> None
         """
         Send an envelope to Sentry.
 
         Envelopes are a data container format that can hold any type of data
         submitted to Sentry. We use it to send all event data (including errors,
-        transactions, crons checkins, etc.) to Sentry.
+        transactions, crons check-ins, etc.) to Sentry.
         """
         pass
 
     def flush(
         self,
-        timeout,  # type: float
-        callback=None,  # type: Optional[Any]
+        timeout,
+        callback=None,
     ):
-        # type: (...) -> None
+        # type: (Self, float, Optional[Any]) -> None
         """
         Wait `timeout` seconds for the current events to be sent out.
 
@@ -127,7 +122,7 @@ def flush(
         return None
 
     def kill(self):
-        # type: () -> None
+        # type: (Self) -> None
         """
         Forcefully kills the transport.
 
@@ -162,19 +157,19 @@ def record_lost_event(
         return None
 
     def is_healthy(self):
-        # type: () -> bool
+        # type: (Self) -> bool
         return True
 
     def __del__(self):
-        # type: () -> None
+        # type: (Self) -> None
         try:
             self.kill()
         except Exception:
             pass
 
 
 def _parse_rate_limits(header, now=None):
-    # type: (Any, Optional[datetime]) -> Iterable[Tuple[Optional[EventDataCategory], datetime]]
+    # type: (str, Optional[datetime]) -> Iterable[Tuple[Optional[EventDataCategory], datetime]]
     if now is None:
         now = datetime.now(timezone.utc)
 
@@ -203,10 +198,8 @@ def _parse_rate_limits(header, now=None):
 class BaseHttpTransport(Transport):
     """The base HTTP transport."""
 
-    def __init__(
-        self, options  # type: Dict[str, Any]
-    ):
-        # type: (...) -> None
+    def __init__(self, options):
+        # type: (Self, Dict[str, Any]) -> None
         from sentry_sdk.consts import VERSION
 
         Transport.__init__(self, options)
@@ -299,11 +292,11 @@ def record_lost_event(
         self._discarded_events[data_category, reason] += quantity
 
     def _get_header_value(self, response, header):
-        # type: (Any, str) -> Optional[str]
+        # type: (Self, Any, str) -> Optional[str]
         return response.headers.get(header)
 
     def _update_rate_limits(self, response):
-        # type: (Union[urllib3.BaseHTTPResponse, httpcore.Response]) -> None
+        # type: (Self, Union[urllib3.BaseHTTPResponse, httpcore.Response]) -> None
 
         # new sentries with more rate limit insights.  We honor this header
         # no matter of the status code to update our internal rate limits.
@@ -329,12 +322,12 @@ def _update_rate_limits(self, response):
 
     def _send_request(
         self,
-        body,  # type: bytes
-        headers,  # type: Dict[str, str]
-        endpoint_type=EndpointType.ENVELOPE,  # type: EndpointType
-        envelope=None,  # type: Optional[Envelope]
+        body,
+        headers,
+        endpoint_type=EndpointType.ENVELOPE,
+        envelope=None,
     ):
-        # type: (...) -> None
+        # type: (Self, bytes, Dict[str, str], EndpointType, Optional[Envelope]) -> None
 
         def record_loss(reason):
             # type: (str) -> None
@@ -384,12 +377,12 @@ def record_loss(reason):
         finally:
             response.close()
 
-    def on_dropped_event(self, reason):
-        # type: (str) -> None
+    def on_dropped_event(self, _reason):
+        # type: (Self, str) -> None
         return None
 
     def _fetch_pending_client_report(self, force=False, interval=60):
-        # type: (bool, int) -> Optional[Item]
+        # type: (Self, bool, int) -> Optional[Item]
         if not self.options["send_client_reports"]:
             return None
 
@@ -420,7 +413,7 @@ def _fetch_pending_client_report(self, force=False, interval=60):
         )
 
     def _flush_client_reports(self, force=False):
-        # type: (bool) -> None
+        # type: (Self, bool) -> None
         client_report = self._fetch_pending_client_report(force=force, interval=60)
         if client_report is not None:
             self.capture_envelope(Envelope(items=[client_report]))
@@ -441,23 +434,21 @@ def _disabled(bucket):
         return _disabled(category) or _disabled(None)
 
     def _is_rate_limited(self):
-        # type: () -> bool
+        # type: (Self) -> bool
         return any(
             ts > datetime.now(timezone.utc) for ts in self._disabled_until.values()
         )
 
     def _is_worker_full(self):
-        # type: () -> bool
+        # type: (Self) -> bool
         return self._worker.full()
 
     def is_healthy(self):
-        # type: () -> bool
+        # type: (Self) -> bool
         return not (self._is_worker_full() or self._is_rate_limited())
 
-    def _send_envelope(
-        self, envelope  # type: Envelope
-    ):
-        # type: (...) -> None
+    def _send_envelope(self, envelope):
+        # type: (Self, Envelope) -> None
 
         # remove all items from the envelope which are over quota
         new_items = []
@@ -510,7 +501,7 @@ def _send_envelope(
         return None
 
     def _serialize_envelope(self, envelope):
-        # type: (..., Envelope) -> tuple[Optional[str], io.BytesIO]
+        # type: (Self, Envelope) -> tuple[Optional[str], io.BytesIO]
         content_encoding = None
         body = io.BytesIO()
         if self._compression_level == 0 or self._compression_algo is None:
@@ -532,11 +523,11 @@ def _serialize_envelope(self, envelope):
         return content_encoding, body
 
     def _get_pool_options(self, ca_certs, cert_file=None, key_file=None):
-        # type: (Optional[Any], Optional[Any], Optional[Any]) -> Dict[str, Any]
+        # type: (Self, Optional[Any], Optional[Any], Optional[Any]) -> Dict[str, Any]
         raise NotImplementedError()
 
     def _in_no_proxy(self, parsed_dsn):
-        # type: (Dsn) -> bool
+        # type: (Self, Dsn) -> bool
         no_proxy = getproxies().get("no")
         if not no_proxy:
             return False
@@ -566,7 +557,7 @@ def _request(
         body,
         headers,
     ):
-        # type: (str, EndpointType, Any, Mapping[str, str]) -> Union[urllib3.BaseHTTPResponse, httpcore.Response]
+        # type: (Self, str, EndpointType, Any, Mapping[str, str]) -> Union[urllib3.BaseHTTPResponse, httpcore.Response]
         raise NotImplementedError()
 
     def capture_envelope(
@@ -586,18 +577,18 @@ def send_envelope_wrapper():
 
     def flush(
         self,
-        timeout,  # type: float
-        callback=None,  # type: Optional[Any]
+        timeout,
+        callback=None,
     ):
-        # type: (...) -> None
+        # type: (Self, float, Optional[Callable]) -> None
         logger.debug("Flushing HTTP transport")
 
         if timeout > 0:
             self._worker.submit(lambda: self._flush_client_reports(force=True))
             self._worker.flush(timeout, callback)
 
     def kill(self):
-        # type: () -> None
+        # type: (Self) -> None
         logger.debug("Killing HTTP transport")
         self._worker.kill()
 
@@ -613,14 +604,14 @@ def _warn_hub_cls():
 
     @property
     def hub_cls(self):
-        # type: () -> type[sentry_sdk.Hub]
+        # type: (Self) -> type[sentry_sdk.Hub]
         """DEPRECATED: This attribute is deprecated and will be removed in a future release."""
         HttpTransport._warn_hub_cls()
         return self._hub_cls
 
     @hub_cls.setter
     def hub_cls(self, value):
-        # type: (type[sentry_sdk.Hub]) -> None
+        # type: (Self, type[sentry_sdk.Hub]) -> None
         """DEPRECATED: This attribute is deprecated and will be removed in a future release."""
         HttpTransport._warn_hub_cls()
         self._hub_cls = value
@@ -631,7 +622,7 @@ class HttpTransport(BaseHttpTransport):
         _pool: Union[PoolManager, ProxyManager]
 
     def _get_pool_options(self, ca_certs, cert_file=None, key_file=None):
-        # type: (Optional[Any], Optional[Any], Optional[Any]) -> Dict[str, Any]
+        # type: (Self, Any, Any, Any) -> Dict[str, Any]
 
         num_pools = self.options.get("_experiments", {}).get("transport_num_pools")
         options = {
@@ -673,9 +664,9 @@ def _make_pool(
         parsed_dsn,  # type: Dsn
         http_proxy,  # type: Optional[str]
         https_proxy,  # type: Optional[str]
-        ca_certs,  # type: Optional[Any]
-        cert_file,  # type: Optional[Any]
-        key_file,  # type: Optional[Any]
+        ca_certs,  # type: Any
+        cert_file,  # type: Any
+        key_file,  # type: Any
         proxy_headers,  # type: Optional[Dict[str, str]]
     ):
         # type: (...) -> Union[PoolManager, ProxyManager]
@@ -724,7 +715,7 @@ def _request(
         body,
         headers,
     ):
-        # type: (str, EndpointType, Any, Mapping[str, str]) -> urllib3.BaseHTTPResponse
+        # type: (Self, str, EndpointType, Any, Mapping[str, str]) -> urllib3.BaseHTTPResponse
         return self._pool.request(
             method,
             self._auth.get_api_url(endpoint_type),
@@ -738,10 +729,8 @@ def _request(
 except ImportError:
     # Sorry, no Http2Transport for you
     class Http2Transport(HttpTransport):
-        def __init__(
-            self, options  # type: Dict[str, Any]
-        ):
-            # type: (...) -> None
+        def __init__(self, options):
+            # type: (Self, Dict[str, Any]) -> None
             super().__init__(options)
             logger.warning(
                 "You tried to use HTTP2Transport but don't have httpcore[http2] installed. Falling back to HTTPTransport."
@@ -758,7 +747,7 @@ class Http2Transport(BaseHttpTransport):  # type: ignore
             ]
 
         def _get_header_value(self, response, header):
-            # type: (httpcore.Response, str) -> Optional[str]
+            # type: (Self, httpcore.Response, str) -> Optional[str]
             return next(
                 (
                     val.decode("ascii")
@@ -775,7 +764,7 @@ def _request(
             body,
             headers,
         ):
-            # type: (str, EndpointType, Any, Mapping[str, str]) -> httpcore.Response
+            # type: (Self, str, EndpointType, Any, Mapping[str, str]) -> httpcore.Response
             response = self._pool.request(
                 method,
                 self._auth.get_api_url(endpoint_type),
@@ -785,7 +774,7 @@ def _request(
             return response
 
         def _get_pool_options(self, ca_certs, cert_file=None, key_file=None):
-            # type: (Optional[Any], Optional[Any], Optional[Any]) -> Dict[str, Any]
+            # type: (Any, Any, Any) -> Dict[str, Any]
             options = {
                 "http2": True,
                 "retries": 3,
@@ -825,9 +814,9 @@ def _make_pool(
             parsed_dsn,  # type: Dsn
             http_proxy,  # type: Optional[str]
             https_proxy,  # type: Optional[str]
-            ca_certs,  # type: Optional[Any]
-            cert_file,  # type: Optional[Any]
-            key_file,  # type: Optional[Any]
+            ca_certs,  # type: Any
+            cert_file,  # type: Any
+            key_file,  # type: Any
             proxy_headers,  # type: Optional[Dict[str, str]]
         ):
             # type: (...) -> Union[httpcore.SOCKSProxy, httpcore.HTTPProxy, httpcore.ConnectionPool]