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,12 +23,22 @@
 
 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
 
 
-def main(client, customer_id, ad_group_id, asset_id):
-    ad_group_asset_service = client.get_service("AdGroupAssetService")
-    ad_group_asset_resource_name = ad_group_asset_service.asset_path(
+def main(
+    client: GoogleAdsClient,
+    customer_id: str,
+    ad_group_id: str,
+    asset_id: str,
+) -> None:
+    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
     )
 
@@ -39,8 +49,10 @@ def main(client, customer_id, ad_group_id, asset_id):
     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,24 +33,32 @@
 
 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"
 
 
-def main(client, customer_id, output_file, write_headers):
+def main(
+    client: GoogleAdsClient,
+    customer_id: str,
+    output_file: str,
+    write_headers: bool,
+) -> None:
     """Writes rows returned from a search_stream request to a CSV file.
     Args:
         client: An initialized GoogleAdsClient instance.
         customer_id (str): The client customer ID string.
         output_file (str): Filename of the file to write the report data to.
         write_headers (bool): From argparse, True if arg is provided.
     """
-    file_dir = os.path.dirname(os.path.abspath(__file__))
-    file_path = os.path.join(file_dir, output_file)
-    ga_service = client.get_service("GoogleAdsService")
+    file_dir: str = os.path.dirname(os.path.abspath(__file__))
+    file_path: str = os.path.join(file_dir, output_file)
+    ga_service: GoogleAdsServiceClient = client.get_service("GoogleAdsService")
 
-    query = """
+    query: str = """
         SELECT
           customer.descriptive_name,
           segments.date,
@@ -65,16 +73,18 @@ def main(client, customer_id, output_file, write_headers):
         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 = [
+        headers: list[str] = [
             "Account",
             "Date",
             "Campaign",

examples/misc/set_custom_client_timeouts.py

@@ -28,36 +28,44 @@
 
 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.
-_QUERY = "SELECT campaign.id FROM campaign"
+_CLIENT_TIMEOUT_SECONDS: int = 5 * 60  # 5 minutes.
+_QUERY: str = "SELECT campaign.id FROM campaign"
 
 
-def main(client, customer_id):
+def main(client: GoogleAdsClient, customer_id: str) -> None:
     """Main method, to run this code example as a standalone application."""
     make_server_streaming_call(client, customer_id)
     make_unary_call(client, customer_id)
 
 
 # [START set_custom_client_timeouts]
-def make_server_streaming_call(client, customer_id):
+def make_server_streaming_call(
+    client: GoogleAdsClient, customer_id: str
+) -> None:
     """Makes a server streaming call using a custom client timeout.
 
     Args:
         client: An initialized GoogleAds client.
         customer_id: The str Google Ads customer ID.
     """
-    ga_service = client.get_service("GoogleAdsService")
-    campaign_ids = []
+    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.
@@ -72,7 +80,7 @@ def make_server_streaming_call(client, customer_id):
                 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:
@@ -92,21 +100,23 @@ def make_server_streaming_call(client, customer_id):
 
 
 # [START set_custom_client_timeouts_1]
-def make_unary_call(client, customer_id):
+def make_unary_call(client: GoogleAdsClient, customer_id: str) -> None:
     """Makes a unary call using a custom client timeout.
 
     Args:
         client: An initialized GoogleAds client.
         customer_id: The Google Ads customer ID.
     """
-    ga_service = client.get_service("GoogleAdsService")
-    campaign_ids = []
+    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
@@ -134,7 +144,7 @@ def make_unary_call(client, customer_id):
             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,20 +23,24 @@
 
 from google.ads.googleads.client import GoogleAdsClient
 from google.ads.googleads.errors import GoogleAdsException
+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
 
 
 # [START upload_image_asset]
-def main(client, customer_id):
+def main(client: GoogleAdsClient, customer_id: str) -> None:
     """Main method, to run this code example as a standalone application."""
 
     # Download image from URL
-    url = "https://gaagl.page.link/Eit5"
-    image_content = get_image_bytes_from_url(url)
+    url: str = "https://gaagl.page.link/Eit5"
+    image_content: bytes = get_image_bytes_from_url(url)
 
-    asset_service = client.get_service("AssetService")
-    asset_operation = client.get_type("AssetOperation")
-    asset = asset_operation.create
+    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
     asset.image_asset.data = image_content
     asset.image_asset.file_size = len(image_content)
@@ -50,8 +54,10 @@ def main(client, customer_id):
     # name, the new name will be dropped silently.
     asset.name = "Marketing Image"
 
-    mutate_asset_response = asset_service.mutate_assets(
-        customer_id=customer_id, operations=[asset_operation]
+    mutate_asset_response: MutateAssetsResponse = (
+        asset_service.mutate_assets(
+            customer_id=customer_id, operations=[asset_operation]
+        )
     )
     print("Uploaded file(s):")
     for row in mutate_asset_response.results: