Finalize type hints in googleads/*.py files

Change-Id: I861a4d620d0214946c17b5f1759cf65053dc1cd5

Author: BenRKarl

google/ads/googleads/client.py

@@ -31,7 +31,7 @@
 )
 
 from types import ModuleType
-from typing import Any, Dict, Tuple, Union
+from typing import Any, Dict, List, Tuple, Union
 
 _logger = logging.getLogger(__name__)
 
@@ -74,7 +74,7 @@ def __init__(self, client: "GoogleAdsClient") -> None:
         """
         self._client: "GoogleAdsClient" = client
         self._version: str = client.version or _DEFAULT_VERSION
-        self._enums: Tuple[str] | None = None
+        self._enums: Union[Tuple[str], None] = None
         self._use_proto_plus: bool = client.use_proto_plus
 
     def __dir__(self) -> Tuple[str]:
@@ -107,7 +107,7 @@ def __getattr__(self, name: str) -> Union[ProtoPlusMessageType, ProtobufMessageT
         try:
             enum_class = self._client.get_type(name)
 
-            if self._use_proto_plus == True:
+            if self._use_proto_plus:
                 for attr in dir(enum_class):
                     attr_val = getattr(enum_class, attr)
                     if isinstance(attr_val, ProtoEnumMeta):
@@ -414,28 +414,28 @@ def get_service(
             options=_GRPC_CHANNEL_OPTIONS,
         )
 
-        interceptors: List[grpc.UnaryUnaryClientInterceptor, grpc.UnaryStreamClientInterceptor] = interceptors + [
-            MetadataInterceptor(  # type: ignore[no-untyped-call]
+        interceptors: List[Union[grpc.UnaryUnaryClientInterceptor, grpc.UnaryStreamClientInterceptor]] = interceptors + [
+            MetadataInterceptor(
                 self.developer_token,
                 self.login_customer_id,
                 self.linked_customer_id,
                 self.use_cloud_org_for_api_access,
             ),
-            LoggingInterceptor(_logger, version, endpoint),  # type: ignore[no-untyped-call]
-            ExceptionInterceptor(  # type: ignore[no-untyped-call]
+            LoggingInterceptor(_logger, version, endpoint),
+            ExceptionInterceptor(
                 version, use_proto_plus=self.use_proto_plus
             ),
         ]
 
-        channel = grpc.intercept_channel(channel, *interceptors)  # type: ignore[no-untyped-call]
+        channel: grpc.Channel = grpc.intercept_channel(channel, *interceptors)
 
-        service_transport = service_transport_class(
+        service_transport: Any = service_transport_class(
             channel=channel, client_info=_CLIENT_INFO
         )
 
         return service_client_class(transport=service_transport)
 
-    def get_type(self, name: str, version: str = _DEFAULT_VERSION):  # type: ignore[no-untyped-def]
+    def get_type(self, name: str, version: str = _DEFAULT_VERSION) -> Union[ProtoPlusMessageType, ProtobufMessageType]:
         """Returns the specified common, enum, error, or resource type.
 
         Args:
@@ -469,19 +469,19 @@ def get_type(self, name: str, version: str = _DEFAULT_VERSION):  # type: ignore[
 
         # If version is specified when the instance is created,
         # override any version specified as an argument.
-        version = self.version if self.version else version
-        type_classes = self._get_api_services_by_version(version)
+        version: str = self.version if self.version else version
+        type_classes: ModuleType = self._get_api_services_by_version(version)
 
         for type_name in _MESSAGE_TYPES:
             if type_name == "services":
-                path = f"{type_name}.types.{name}"
+                path: str = f"{type_name}.types.{name}"
             else:
-                path = f"{type_name}.{name}"
+                path: str = f"{type_name}.{name}"
 
             try:
-                message_class = util.get_nested_attr(type_classes, path)  # type: ignore[no-untyped-call]
+                message_class: Union[ProtoPlusMessageType, ProtobufMessageType] = util.get_nested_attr(type_classes, path)  # type: ignore[no-untyped-call]
 
-                if self.use_proto_plus == True:
+                if self.use_proto_plus:
                     return message_class()
                 else:
                     return util.convert_proto_plus_to_protobuf(message_class())  # type: ignore[no-untyped-call]

google/ads/googleads/config.py

@@ -18,6 +18,7 @@
 import logging.config
 import os
 import re
+from typing import Any, Callable, List, Tuple, TypeVar, Union
 import yaml
 
 
@@ -45,7 +46,6 @@
     "path_to_private_key_file",
     "delegated_account",
 )
-from typing import Any, Callable, TypeVar
 
 _KEYS_ENV_VARIABLES_MAP: dict[str, str] = {
     key: _ENV_PREFIX + key.upper()
@@ -71,12 +71,12 @@ def _config_validation_decorator(func: F) -> F:
     """
 
     @functools.wraps(func)
-    def validation_wrapper(*args: Any, **kwargs: Any) -> Any:
+    def validation_wrapper(*args: Any, **kwargs: Any) -> dict[str, Any]:
         config_dict: dict[str, Any] = func(*args, **kwargs)
         validate_dict(config_dict)
         return config_dict
 
-    return validation_wrapper  # type: ignore[return-value]
+    return validation_wrapper
 
 
 def _config_parser_decorator(func: F) -> F:
@@ -94,17 +94,17 @@ def parser_wrapper(*args: Any, **kwargs: Any) -> Any:
         parsed_config: dict[str, Any] = convert_login_customer_id_to_str(
             config_dict
         )
-        parsed_config = convert_linked_customer_id_to_str(parsed_config)
+        parsed_config: dict[str, Any] = convert_linked_customer_id_to_str(parsed_config)
 
-        config_keys = parsed_config.keys()
+        config_keys: List[str] = parsed_config.keys()
 
         if "logging" in config_keys:
-            logging_config = parsed_config["logging"]
+            logging_config: dict[str, Any] = parsed_config["logging"]
             # If the logging config is a dict then it is already in the format
             # that needs to be returned by this method.
             if type(logging_config) is not dict:
                 try:
-                    parsed_config["logging"] = json.loads(logging_config)
+                    parsed_config["logging"]: dict[str, Any] = json.loads(logging_config)
                     # The logger is configured here in case deprecation warnings
                     # need to be logged further down in this method. The logger
                     # is otherwise configured by the GoogleAdsClient class.
@@ -156,12 +156,12 @@ def parser_wrapper(*args: Any, **kwargs: Any) -> Any:
             # variable we need to manually change it to the bool False because
             # the string "False" is truthy and can easily be incorrectly
             # converted to the boolean True.
-            value = parsed_config.get("use_proto_plus", False)
-            parsed_config["use_proto_plus"] = disambiguate_string_bool(value)
+            value: Union[str, bool] = parsed_config.get("use_proto_plus", False)
+            parsed_config["use_proto_plus"]: bool = disambiguate_string_bool(value)
 
         return parsed_config
 
-    return parser_wrapper  # type: ignore[return-value]
+    return parser_wrapper
 
 
 def validate_dict(config_data: dict[str, Any]) -> None:
@@ -178,7 +178,7 @@ def validate_dict(config_data: dict[str, Any]) -> None:
     Raises:
         ValueError: If the dict does not contain all required config keys.
     """
-    if not "use_proto_plus" in config_data.keys():
+    if "use_proto_plus" not in config_data.keys():
         raise ValueError(
             "The client library configuration is missing the required "
             '"use_proto_plus" key. Please set this option to either "True" '
@@ -200,7 +200,7 @@ def validate_dict(config_data: dict[str, Any]) -> None:
         validate_linked_customer_id(str(config_data["linked_customer_id"]))
 
 
-def _validate_customer_id(customer_id: str | None, id_type: str) -> None:
+def _validate_customer_id(customer_id: Union[str, None], id_type: str) -> None:
     """Validates a customer ID.
 
     Args:
@@ -214,15 +214,15 @@ def _validate_customer_id(customer_id: str | None, id_type: str) -> None:
     """
     if customer_id is not None:
         # Checks that the string is comprised only of 10 digits.
-        pattern = re.compile(r"^\d{10}", re.ASCII)
+        pattern: re.Pattern = re.compile(r"^\d{10}", re.ASCII)
         if not pattern.fullmatch(customer_id):
             raise ValueError(
                 f"The specified {id_type} customer ID is invalid. It must be a "
                 "ten digit number represented as a string, i.e. '1234567890'"
             )
 
 
-def validate_login_customer_id(login_customer_id: str | None) -> None:
+def validate_login_customer_id(login_customer_id: Union[str, None]) -> None:
     """Validates a login customer ID.
     Args:
         login_customer_id: a str from config indicating a login customer ID.
@@ -233,7 +233,7 @@ def validate_login_customer_id(login_customer_id: str | None) -> None:
     _validate_customer_id(login_customer_id, "login")
 
 
-def validate_linked_customer_id(linked_customer_id: str | None) -> None:
+def validate_linked_customer_id(linked_customer_id: Union[str, None]) -> None:
     """Validates a linked customer ID.
     Args:
         linked_customer_id: a str from config indicating a linked customer ID.
@@ -246,7 +246,7 @@ def validate_linked_customer_id(linked_customer_id: str | None) -> None:
 
 @_config_validation_decorator
 @_config_parser_decorator
-def load_from_yaml_file(path: str | None = None) -> dict[str, Any]:
+def load_from_yaml_file(path: Union[str, None] = None) -> dict[str, Any]:
     """Loads configuration data from a YAML file and returns it as a dict.
 
     Args:
@@ -264,20 +264,20 @@ def load_from_yaml_file(path: str | None = None) -> dict[str, Any]:
         # If no path is specified then we check for the environment variable
         # that may define the path. If that is not defined then we use the
         # default path.
-        path_from_env_var = os.environ.get(
+        path_from_env_var: str = os.environ.get(
             _ENV_PREFIX + _CONFIG_FILE_PATH_KEY[0].upper()
         )
-        path = (
+        path: Tuple[str] = (
             path_from_env_var
             if path_from_env_var
             else os.path.join(os.path.expanduser("~"), "google-ads.yaml")
         )
 
     if not os.path.isabs(path):
-        path = os.path.expanduser(path)
+        path: str = os.path.expanduser(path)
 
     with open(path, "rb") as handle:
-        yaml_doc = handle.read()
+        yaml_doc: bytes = handle.read()
 
     return parse_yaml_document_to_dict(yaml_doc)
 
@@ -308,7 +308,7 @@ def load_from_dict(config_dict: dict[str, Any]) -> dict[str, Any]:
 
 @_config_validation_decorator
 @_config_parser_decorator
-def parse_yaml_document_to_dict(yaml_doc: str | bytes) -> dict[str, Any]:
+def parse_yaml_document_to_dict(yaml_doc: Union[str, bytes]) -> dict[str, Any]:
     """Parses a YAML document to a dict.
 
     Args:
@@ -382,10 +382,10 @@ def convert_login_customer_id_to_str(
     Returns:
         The same config dict object with a mutated login_customer_id attr.
     """
-    login_customer_id: Any = config_data.get("login_customer_id")
+    login_customer_id: str = config_data.get("login_customer_id")
 
     if login_customer_id:
-        config_data["login_customer_id"] = str(login_customer_id)
+        config_data["login_customer_id"]: str = str(login_customer_id)
 
     return config_data
 
@@ -405,15 +405,15 @@ def convert_linked_customer_id_to_str(
     Returns:
         The same config dict object with a mutated linked_customer_id attr.
     """
-    linked_customer_id: Any = config_data.get("linked_customer_id")
+    linked_customer_id: str = config_data.get("linked_customer_id")
 
     if linked_customer_id:
-        config_data["linked_customer_id"] = str(linked_customer_id)
+        config_data["linked_customer_id"]: str = str(linked_customer_id)
 
     return config_data
 
 
-def disambiguate_string_bool(value: str | bool) -> bool:
+def disambiguate_string_bool(value: Union[str, bool]) -> bool:
     """Converts a stringified boolean to its bool representation.
 
     Args:

google/ads/googleads/errors.py

@@ -13,9 +13,8 @@
 # limitations under the License.
 """Errors used by the Google Ads API library."""
 
-
-from typing import Any
 import grpc
+from proto import Message as ProtobufMessageType
 
 
 class GoogleAdsException(Exception):
@@ -25,7 +24,7 @@ def __init__(
         self,
         error: grpc.RpcError,
         call: grpc.Call,
-        failure: Any,  # Replace Any with GoogleAdsFailure when available
+        failure: ProtobufMessageType,
         request_id: str,
     ) -> None:
         """Initializer.
@@ -39,5 +38,5 @@ def __init__(
         """
         self.error: grpc.RpcError = error
         self.call: grpc.Call = call
-        self.failure: Any = failure  # Replace Any with GoogleAdsFailure
+        self.failure: ProtobufMessageType = failure
         self.request_id: str = request_id

google/ads/googleads/oauth2.py

@@ -23,7 +23,7 @@
 
 from google.ads.googleads import config
 
-from typing import Any, Callable, TypeVar
+from typing import Any, Callable, TypeVar, Union
 
 _SERVICE_ACCOUNT_SCOPES: list[str] = [
     "https://www.googleapis.com/auth/adwords"
@@ -42,27 +42,27 @@ def _initialize_credentials_decorator(func: F) -> F:
 
     @functools.wraps(func)
     def initialize_credentials_wrapper(*args: Any, **kwargs: Any) -> Any:
-        credentials = func(*args, **kwargs)
+        credentials: Union[InstalledAppCredentials, ServiceAccountCreds] = func(*args, **kwargs)
         # If the configs contain an http_proxy, refresh credentials through the
         # proxy URI
-        proxy: str | None = kwargs.get("http_proxy")
+        proxy: Union[str, None] = kwargs.get("http_proxy")
         if proxy:
-            session = Session()
+            session: Session = Session()
             session.proxies.update({"http": proxy, "https": proxy})
             credentials.refresh(Request(session=session))
         else:
-            credentials.refresh(Request())  # type: ignore[no-untyped-call]
+            credentials.refresh(Request())
         return credentials
 
-    return initialize_credentials_wrapper  # type: ignore[return-value]
+    return initialize_credentials_wrapper
 
 
 @_initialize_credentials_decorator
 def get_installed_app_credentials(
     client_id: str,
     client_secret: str,
     refresh_token: str,
-    http_proxy: str | None = None,
+    http_proxy: Union[str, None] = None,
     token_uri: str = _DEFAULT_TOKEN_URI,
 ) -> InstalledAppCredentials:
     """Creates and returns an instance of oauth2.credentials.Credentials.
@@ -90,7 +90,7 @@ def get_installed_app_credentials(
 def get_service_account_credentials(
     json_key_file_path: str,
     subject: str,
-    http_proxy: str | None = None,
+    http_proxy: Union[str, None] = None,
     scopes: list[str] = _SERVICE_ACCOUNT_SCOPES,
 ) -> ServiceAccountCreds:
     """Creates and returns an instance of oauth2.service_account.Credentials.
@@ -111,7 +111,7 @@ def get_service_account_credentials(
 
 def get_credentials(
     config_data: dict[str, Any]
-) -> InstalledAppCredentials | ServiceAccountCreds:
+) -> Union[InstalledAppCredentials, ServiceAccountCreds]:
     """Decides which type of credentials to return based on the given config.
 
     Args: