Add type hints and annotations to examples/misc

This change adds type hints and annotations to the Python files in the examples/misc directory to improve code readability and maintainability.

Author: google-labs-jules[bot]

examples/misc/add_ad_group_image_asset.py

@@ -23,6 +23,9 @@
 
 from google.ads.googleads.client import GoogleAdsClient
 from google.ads.googleads.errors import GoogleAdsException
+from google.ads.googleads.v19.services.types.ad_group_asset_service import AdGroupAssetServiceClient
+from google.ads.googleads.v19.services.types.ad_group_asset_service import MutateAdGroupAssetsResponse
+from google.ads.googleads.v19.services.types.google_ads_service import GoogleAdsServiceClient
 from google.api_core import protobuf_helpers
 
 
@@ -32,7 +35,9 @@ def main(
     ad_group_id: str,
     asset_id: str,
 ) -> None:
-    ad_group_asset_service = client.get_service("AdGroupAssetService")
+    ad_group_asset_service: AdGroupAssetServiceClient = client.get_service(
+        "AdGroupAssetService"
+    )
     ad_group_asset_resource_name: str = ad_group_asset_service.asset_path(
         customer_id, asset_id
     )
@@ -44,8 +49,10 @@ def main(
     ad_group_asset_set.ad_group = ad_group_asset_service.ad_group_path(
         customer_id, ad_group_id
     )
-    response = ad_group_asset_service.mutate_ad_group_assets(
-        customer_id=customer_id, operations=[ad_group_asset_operation]
+    response: MutateAdGroupAssetsResponse = (
+        ad_group_asset_service.mutate_ad_group_assets(
+            customer_id=customer_id, operations=[ad_group_asset_operation]
+        )
     )
 
     for result in response.results:

examples/misc/campaign_report_to_csv.py

@@ -33,6 +33,9 @@
 
 from google.ads.googleads.client import GoogleAdsClient
 from google.ads.googleads.errors import GoogleAdsException
+from google.ads.googleads.v19.services.services.google_ads_service import GoogleAdsServiceClient
+from google.ads.googleads.v19.services.types.google_ads_service import SearchGoogleAdsStreamRequest
+from google.api_core.call import GrpcStream
 
 
 _DEFAULT_FILE_NAME = "campaign_report_to_csv_results.csv"
@@ -53,7 +56,7 @@ def main(
     """
     file_dir: str = os.path.dirname(os.path.abspath(__file__))
     file_path: str = os.path.join(file_dir, output_file)
-    ga_service = client.get_service("GoogleAdsService")
+    ga_service: GoogleAdsServiceClient = client.get_service("GoogleAdsService")
 
     query: str = """
         SELECT
@@ -70,13 +73,15 @@ def main(
         LIMIT 25"""
 
     # Issues a search request using streaming.
-    search_request = client.get_type("SearchGoogleAdsStreamRequest")
+    search_request: SearchGoogleAdsStreamRequest = client.get_type(
+        "SearchGoogleAdsStreamRequest"
+    )
     search_request.customer_id = customer_id
     search_request.query = query
-    stream = ga_service.search_stream(search_request)
+    stream: GrpcStream = ga_service.search_stream(search_request)
 
     with open(file_path, "w", newline="") as f:
-        writer = csv.writer(f)
+        writer: csv.writer = csv.writer(f)
 
         # Define a list of headers for the first row.
         headers: list[str] = [

examples/misc/set_custom_client_timeouts.py

@@ -29,11 +29,15 @@
 
 from google.ads.googleads.client import GoogleAdsClient
 from google.ads.googleads.errors import GoogleAdsException
+from google.ads.googleads.v19.services.services.google_ads_service import GoogleAdsServiceClient
+from google.ads.googleads.v19.services.types.google_ads_service import SearchGoogleAdsRequest
+from google.ads.googleads.v19.services.types.google_ads_service import SearchGoogleAdsStreamRequest
+from google.ads.googleads.v19.services.types.google_ads_service import SearchGoogleAdsStreamResponse
 from google.api_core.exceptions import DeadlineExceeded
 from google.api_core.retry import Retry
 
 
-_CLIENT_TIMEOUT_SECONDS = 5 * 60  # 5 minutes.
+_CLIENT_TIMEOUT_SECONDS: int = 5 * 60  # 5 minutes.
 _QUERY: str = "SELECT campaign.id FROM campaign"
 
 
@@ -53,14 +57,16 @@ def make_server_streaming_call(
         client: An initialized GoogleAds client.
         customer_id: The str Google Ads customer ID.
     """
-    ga_service = client.get_service("GoogleAdsService")
-    campaign_ids: List[str] = []
+    ga_service: GoogleAdsServiceClient = client.get_service("GoogleAdsService")
+    campaign_ids: list[str] = []
 
     try:
-        search_request = client.get_type("SearchGoogleAdsStreamRequest")
+        search_request: SearchGoogleAdsStreamRequest = client.get_type(
+            "SearchGoogleAdsStreamRequest"
+        )
         search_request.customer_id = customer_id
         search_request.query = _QUERY
-        stream = ga_service.search_stream(
+        stream: SearchGoogleAdsStreamResponse = ga_service.search_stream(
             request=search_request,
             # When making any request, an optional "timeout" parameter can be
             # provided to specify a client-side response deadline in seconds.
@@ -75,7 +81,7 @@ def make_server_streaming_call(
                 campaign_ids.append(row.campaign.id)
 
         print("The server streaming call completed before the timeout.")
-    except DeadlineExceeded as ex:
+    except DeadlineExceeded:
         print("The server streaming call did not complete before the timeout.")
         sys.exit(1)
     except GoogleAdsException as ex:
@@ -102,14 +108,16 @@ def make_unary_call(client: GoogleAdsClient, customer_id: str) -> None:
         client: An initialized GoogleAds client.
         customer_id: The Google Ads customer ID.
     """
-    ga_service = client.get_service("GoogleAdsService")
-    campaign_ids: List[str] = []
+    ga_service: GoogleAdsServiceClient = client.get_service("GoogleAdsService")
+    campaign_ids: list[str] = []
 
     try:
-        search_request = client.get_type("SearchGoogleAdsRequest")
+        search_request: SearchGoogleAdsRequest = client.get_type(
+            "SearchGoogleAdsRequest"
+        )
         search_request.customer_id = customer_id
         search_request.query = _QUERY
-        results = ga_service.search(
+        results: SearchGoogleAdsStreamResponse = ga_service.search(
             request=search_request,
             # When making any request, an optional "retry" parameter can be
             # provided to specify its retry behavior. Complete information about
@@ -137,7 +145,7 @@ def make_unary_call(client: GoogleAdsClient, customer_id: str) -> None:
             campaign_ids.append(row.campaign.id)
 
         print("The unary call completed before the timeout.")
-    except DeadlineExceeded as ex:
+    except DeadlineExceeded:
         print("The unary call did not complete before the timeout.")
         sys.exit(1)
     except GoogleAdsException as ex:

examples/misc/upload_image_asset.py

@@ -23,11 +23,10 @@
 
 from google.ads.googleads.client import GoogleAdsClient
 from google.ads.googleads.errors import GoogleAdsException
-from google.ads.googleads.v19.services.types import (
-    AssetOperation,
-    MutateAssetsResponse,
-)
-from google.ads.googleads.v19.resources.types import Asset
+from google.ads.googleads.v19.services.services.asset_service import AssetServiceClient
+from google.ads.googleads.v19.services.types.asset_service import MutateAssetsResponse
+from google.ads.googleads.v19.resources.types.asset import Asset
+from google.ads.googleads.v19.services.types.asset_service import AssetOperation
 from examples.utils.example_helpers import get_image_bytes_from_url
 
 
@@ -39,7 +38,7 @@ def main(client: GoogleAdsClient, customer_id: str) -> None:
     url: str = "https://gaagl.page.link/Eit5"
     image_content: bytes = get_image_bytes_from_url(url)
 
-    asset_service = client.get_service("AssetService")
+    asset_service: AssetServiceClient = client.get_service("AssetService")
     asset_operation: AssetOperation = client.get_type("AssetOperation")
     asset: Asset = asset_operation.create
     asset.type_ = client.enums.AssetTypeEnum.IMAGE
@@ -54,6 +53,7 @@ def main(client: GoogleAdsClient, customer_id: str) -> None:
     # When there is an existing image asset with the same content but a different
     # name, the new name will be dropped silently.
     asset.name = "Marketing Image"
+
     mutate_asset_response: MutateAssetsResponse = (
         asset_service.mutate_assets(
             customer_id=customer_id, operations=[asset_operation]