Add type annotations to interceptor and fix style issues in helpers

Change-Id: I6f570470fada9980b1486dfda77201b4e865f701

Author: BenRKarl

google/ads/googleads/interceptors/helpers.py

@@ -16,7 +16,7 @@
 from typing import Any, Dict, List, Tuple, Union
 
 from google.protobuf.descriptor import FieldDescriptor
-from google.protobuf.message import ProtobufMessage
+from google.protobuf.message import Message as ProtobufMessage
 from proto import Message as ProtoPlusMessage
 
 from google.ads.googleads.util import (

google/ads/googleads/interceptors/interceptor.py

@@ -20,33 +20,46 @@
 
 from dataclasses import dataclass
 from importlib import import_module
-from typing import AnyStr, Optional, Sequence, Tuple
+from types import ModuleType
+from typing import (
+    Any,
+    AnyStr,
+    Dict,
+    List,
+    Optional,
+    Sequence,
+    Tuple
+)
 import json
 
-from google.protobuf.message import DecodeError
-from grpc import ClientCallDetails, StatusCode, CallCredentials
+from google.protobuf.message import DecodeError, Message
+from grpc import Call, ClientCallDetails, StatusCode, CallCredentials, RpcError
 
 from google.ads.googleads.errors import GoogleAdsException
 
-_REQUEST_ID_KEY = "request-id"
+_REQUEST_ID_KEY: str = "request-id"
 # Codes that are retried upon by google.api_core.
-_RETRY_STATUS_CODES = (StatusCode.INTERNAL, StatusCode.RESOURCE_EXHAUSTED)
+_RETRY_STATUS_CODES: Tuple[StatusCode]= (StatusCode.INTERNAL, StatusCode.RESOURCE_EXHAUSTED)
+
+Metadata = Sequence[Tuple[str, AnyStr]]
 
 
 class Interceptor:
-    _SENSITIVE_INFO_MASK = "REDACTED"
+    _SENSITIVE_INFO_MASK: str = "REDACTED"
 
     @dataclass
     class _ClientCallDetails(ClientCallDetails):
         """Wrapper class for initializing a new ClientCallDetails instance."""
 
         method: str
         timeout: Optional[float]
-        metadata: Optional[Sequence[Tuple[str, AnyStr]]]
+        metadata: Optional[Metadata]
         credentials: Optional[CallCredentials]
 
     @classmethod
-    def get_request_id_from_metadata(cls, trailing_metadata):
+    def get_request_id_from_metadata(
+        cls, trailing_metadata: Metadata
+    ) -> Optional[str]:
         """Gets the request ID for the Google Ads API request.
 
         Args:
@@ -58,12 +71,12 @@ def get_request_id_from_metadata(cls, trailing_metadata):
         """
         for kv in trailing_metadata:
             if kv[0] == _REQUEST_ID_KEY:
-                return kv[1]  # Return the found request ID.
+                return kv[1]
 
         return None
 
     @classmethod
-    def parse_metadata_to_json(cls, metadata):
+    def parse_metadata_to_json(cls, metadata: Optional[Metadata]) -> str:
         """Parses metadata from gRPC request and response messages to a JSON str.
 
         Obscures the value for "developer-token".
@@ -74,23 +87,23 @@ def parse_metadata_to_json(cls, metadata):
         Returns:
             A str of metadata formatted as JSON key/value pairs.
         """
-        metadata_dict = {}
+        metadata_dict: Dict = {}
 
         if metadata is None:
             return "{}"
 
         for datum in metadata:
-            key = datum[0]
+            key: str = datum[0]
             if key == "developer-token":
                 metadata_dict[key] = cls._SENSITIVE_INFO_MASK
             else:
-                value = datum[1]
+                value: AnyStr = datum[1]
                 metadata_dict[key] = value
 
         return cls.format_json_object(metadata_dict)
 
     @classmethod
-    def format_json_object(cls, obj):
+    def format_json_object(cls, obj: Dict) -> str:
         """Parses a serializable object into a consistently formatted JSON string.
 
         Returns:
@@ -102,8 +115,7 @@ def format_json_object(cls, obj):
         Returns:
             A str of metadata formatted as JSON key/value pairs.
         """
-
-        def default_serializer(value):
+        def default_serializer(value: Any) -> Optional[str]:
             if isinstance(value, bytes):
                 return value.decode(errors="ignore")
             else:
@@ -121,7 +133,9 @@ def default_serializer(value):
         )
 
     @classmethod
-    def get_trailing_metadata_from_interceptor_exception(cls, exception):
+    def get_trailing_metadata_from_interceptor_exception(
+        cls, exception: RpcError
+    ) -> Metadata:
         """Retrieves trailing metadata from an exception object.
 
         Args:
@@ -146,8 +160,12 @@ def get_trailing_metadata_from_interceptor_exception(cls, exception):
 
     @classmethod
     def get_client_call_details_instance(
-        cls, method, timeout, metadata, credentials=None
-    ):
+        cls,
+        method: str,
+        timeout: float,
+        metadata: List[Tuple[str, AnyStr]],
+        credentials: Optional[CallCredentials] = None,
+    ) -> ClientCallDetails:
         """Initializes an instance of the ClientCallDetails with the given data.
 
         Args:
@@ -161,14 +179,14 @@ def get_client_call_details_instance(
         """
         return cls._ClientCallDetails(method, timeout, metadata, credentials)
 
-    def __init__(self, api_version):
-        self._error_protos = None
-        self._failure_key = (
+    def __init__(self, api_version: str):
+        self._error_protos: Optional[ModuleType] = None
+        self._failure_key: str = (
             f"google.ads.googleads.{api_version}.errors.googleadsfailure-bin"
         )
-        self._api_version = api_version
+        self._api_version: str = api_version
 
-    def _get_error_from_response(self, response):
+    def _get_error_from_response(self, response: Call) -> Call:
         """Attempts to wrap failed responses as GoogleAdsException instances.
 
         Handles failed gRPC responses of by attempting to convert them
@@ -192,15 +210,17 @@ def _get_error_from_response(self, response):
             Exception: If not a GoogleAdsException or RpcException the error
                 will be raised as-is.
         """
-        status_code = response.code()
-        response_exception = response.exception()
+        status_code: StatusCode = response.code()
+        response_exception: Call = response.exception()
 
         if status_code not in _RETRY_STATUS_CODES:
-            trailing_metadata = response.trailing_metadata()
-            google_ads_failure = self._get_google_ads_failure(trailing_metadata)
+            trailing_metadata: Metadata = response.trailing_metadata()
+            google_ads_failure: Optional[Message] = self._get_google_ads_failure(
+                trailing_metadata
+            )
 
-            if google_ads_failure:
-                request_id = self.get_request_id_from_metadata(
+            if google_ads_failure and response_exception:
+                request_id: Optional[str] = self.get_request_id_from_metadata(
                     trailing_metadata
                 )
 
@@ -211,17 +231,18 @@ def _get_error_from_response(self, response):
                 return GoogleAdsException(
                     response_exception, response, google_ads_failure, request_id
                 )
-            else:
+            elif response_exception:
                 # Raise the original exception if not a GoogleAdsFailure. This
                 # type of error is generally caused by problems at the request
                 # level, such as when an invalid endpoint is given.
                 return response_exception
-        else:
+        elif response_exception:
             # Raise the original exception if error has status code
             # INTERNAL or RESOURCE_EXHAUSTED, meaning that
             return response_exception
+        raise response.exception()
 
-    def _get_google_ads_failure(self, trailing_metadata):
+    def _get_google_ads_failure(self, trailing_metadata: Metadata) -> Optional[Message]:
         """Gets the Google Ads failure details if they exist.
 
         Args: