Add annotations to examples/error_handling (#948)

Author: bobhancockg

examples/error_handling/handle_keyword_policy_violations.py

@@ -27,12 +27,25 @@
 
 import argparse
 import sys
+from typing import Any, List, Optional, Tuple
 
 from google.ads.googleads.client import GoogleAdsClient
 from google.ads.googleads.errors import GoogleAdsException
-
-
-def main(client, customer_id, ad_group_id, keyword_text):
+from google.ads.googleads.v20.services.services.ad_group_criterion_service import (
+    AdGroupCriterionServiceClient,
+)
+from google.ads.googleads.v20.services.types.ad_group_criterion_service import (
+    AdGroupCriterionOperation,
+)
+from google.ads.googleads.v20.common.types.policy import PolicyViolationKey
+
+
+def main(
+    client: GoogleAdsClient,
+    customer_id: str,
+    ad_group_id: str,
+    keyword_text: str,
+) -> None:
     """Demonstrates how to request an exemption for keyword policy violations.
 
     Args:
@@ -42,8 +55,12 @@ def main(client, customer_id, ad_group_id, keyword_text):
         keyword_text: The keyword text to add.
     """
 
-    ad_group_criterion_service = client.get_service("AdGroupCriterionService")
+    ad_group_criterion_service: AdGroupCriterionServiceClient = (
+        client.get_service("AdGroupCriterionService")
+    )
 
+    googleads_exception: Optional[GoogleAdsException]
+    ad_group_criterion_operation: AdGroupCriterionOperation
     (
         googleads_exception,
         ad_group_criterion_operation,
@@ -60,8 +77,8 @@ def main(client, customer_id, ad_group_id, keyword_text):
         # your keyword contains many policy violations, but not all of them are
         # exemptible, the request will not be sent.
         if googleads_exception is not None:
-            exempt_policy_violation_keys = fetch_exempt_policy_violation_keys(
-                googleads_exception
+            exempt_policy_violation_keys: List[PolicyViolationKey] = (
+                fetch_exempt_policy_violation_keys(googleads_exception)
             )
             request_exemption(
                 customer_id,
@@ -83,8 +100,12 @@ def main(client, customer_id, ad_group_id, keyword_text):
 
 
 def create_keyword_criterion(
-    client, ad_group_criterion_service, customer_id, ad_group_id, keyword_text
-):
+    client: GoogleAdsClient,
+    ad_group_criterion_service: AdGroupCriterionServiceClient,
+    customer_id: str,
+    ad_group_id: str,
+    keyword_text: str,
+) -> Tuple[Optional[GoogleAdsException], AdGroupCriterionOperation]:
     """Attempts to add a keyword criterion to an ad group.
 
     Args:
@@ -99,8 +120,10 @@ def create_keyword_criterion(
         successful) and the modified operation.
     """
     # Constructs an ad group criterion using the keyword text provided.
-    ad_group_criterion_operation = client.get_type("AdGroupCriterionOperation")
-    ad_group_criterion = ad_group_criterion_operation.create
+    ad_group_criterion_operation: AdGroupCriterionOperation = client.get_type(
+        "AdGroupCriterionOperation"
+    )
+    ad_group_criterion: Any = ad_group_criterion_operation.create
     ad_group_criterion.ad_group = client.get_service(
         "AdGroupService"
     ).ad_group_path(customer_id, ad_group_id)
@@ -112,7 +135,7 @@ def create_keyword_criterion(
 
     try:
         # Try sending a mutate request to add the keyword.
-        response = ad_group_criterion_service.mutate_ad_group_criteria(
+        response: Any = ad_group_criterion_service.mutate_ad_group_criteria(
             customer_id=customer_id, operations=[ad_group_criterion_operation]
         )
     except GoogleAdsException as googleads_exception:
@@ -129,7 +152,9 @@ def create_keyword_criterion(
 
 
 # [START handle_keyword_policy_violations]
-def fetch_exempt_policy_violation_keys(googleads_exception):
+def fetch_exempt_policy_violation_keys(
+    googleads_exception: GoogleAdsException,
+) -> List[PolicyViolationKey]:
     """Collects all policy violation keys that can be exempted.
 
     Args:
@@ -138,7 +163,7 @@ def fetch_exempt_policy_violation_keys(googleads_exception):
     Returns:
         A list of policy violation keys.
     """
-    exempt_policy_violation_keys = []
+    exempt_policy_violation_keys: List[PolicyViolationKey] = []
 
     print("Google Ads failure details:")
     for error in googleads_exception.failure.errors:
@@ -191,11 +216,11 @@ def fetch_exempt_policy_violation_keys(googleads_exception):
 
 # [START handle_keyword_policy_violations_1]
 def request_exemption(
-    customer_id,
-    ad_group_criterion_service,
-    ad_group_criterion_operation,
-    exempt_policy_violation_keys,
-):
+    customer_id: str,
+    ad_group_criterion_service: AdGroupCriterionServiceClient,
+    ad_group_criterion_operation: AdGroupCriterionOperation,
+    exempt_policy_violation_keys: List[PolicyViolationKey],
+) -> None:
     """Sends exemption requests for creating a keyword.
 
     Args:
@@ -212,7 +237,7 @@ def request_exemption(
     ad_group_criterion_operation.exempt_policy_violation_keys.extend(
         exempt_policy_violation_keys
     )
-    response = ad_group_criterion_service.mutate_ad_group_criteria(
+    response: Any = ad_group_criterion_service.mutate_ad_group_criteria(
         customer_id=customer_id, operations=[ad_group_criterion_operation]
     )
     print(
@@ -256,7 +281,9 @@ def request_exemption(
 
     # GoogleAdsClient will read the google-ads.yaml configuration file in the
     # home directory if none is specified.
-    googleads_client = GoogleAdsClient.load_from_storage(version="v20")
+    googleads_client: GoogleAdsClient = GoogleAdsClient.load_from_storage(
+        version="v20"
+    )
 
     main(
         googleads_client, args.customer_id, args.ad_group_id, args.keyword_text

examples/error_handling/handle_partial_failure.py

@@ -18,12 +18,24 @@
 import argparse
 import sys
 import uuid
+from typing import Any, List
 
 from google.ads.googleads.client import GoogleAdsClient
 from google.ads.googleads.errors import GoogleAdsException
-
-
-def main(client, customer_id, campaign_id):
+from google.ads.googleads.v20.services.services.ad_group_service import (
+    AdGroupServiceClient,
+)
+from google.ads.googleads.v20.services.services.campaign_service import (
+    CampaignServiceClient,
+)
+from google.ads.googleads.v20.services.types.ad_group_service import (
+    AdGroupOperation,
+    MutateAdGroupsResponse,
+    MutateAdGroupsRequest,
+)
+
+
+def main(client: GoogleAdsClient, customer_id: str, campaign_id: str) -> None:
     """Runs the example code, which demonstrates how to handle partial failures.
 
     The example creates three Ad Groups, two of which intentionally fail in
@@ -36,7 +48,9 @@ def main(client, customer_id, campaign_id):
         campaign_id: The ID for a campaign to create Ad Groups under.
     """
     try:
-        ad_group_response = create_ad_groups(client, customer_id, campaign_id)
+        ad_group_response: MutateAdGroupsResponse = create_ad_groups(
+            client, customer_id, campaign_id
+        )
     except GoogleAdsException as ex:
         print(
             f'Request with ID "{ex.request_id}" failed with status '
@@ -53,7 +67,9 @@ def main(client, customer_id, campaign_id):
 
 
 # [START handle_partial_failure]
-def create_ad_groups(client, customer_id, campaign_id):
+def create_ad_groups(
+    client: GoogleAdsClient, customer_id: str, campaign_id: str
+) -> MutateAdGroupsResponse:
     """Creates three Ad Groups, two of which intentionally generate errors.
 
     Args:
@@ -63,35 +79,41 @@ def create_ad_groups(client, customer_id, campaign_id):
 
     Returns: A MutateAdGroupsResponse message instance.
     """
-    ad_group_service = client.get_service("AdGroupService")
-    campaign_service = client.get_service("CampaignService")
-    resource_name = campaign_service.campaign_path(customer_id, campaign_id)
+    ad_group_service: AdGroupServiceClient = client.get_service(
+        "AdGroupService"
+    )
+    campaign_service: CampaignServiceClient = client.get_service(
+        "CampaignService"
+    )
+    resource_name: str = campaign_service.campaign_path(
+        customer_id, campaign_id
+    )
 
-    invalid_resource_name = campaign_service.campaign_path(customer_id, 0)
-    ad_group_operations = []
+    invalid_resource_name: str = campaign_service.campaign_path(customer_id, 0)
+    ad_group_operations: List[AdGroupOperation] = []
 
     # This AdGroup should be created successfully - assuming the campaign in
     # the params exists.
-    ad_group_op1 = client.get_type("AdGroupOperation")
+    ad_group_op1: AdGroupOperation = client.get_type("AdGroupOperation")
     ad_group_op1.create.name = f"Valid AdGroup: {uuid.uuid4()}"
     ad_group_op1.create.campaign = resource_name
     ad_group_operations.append(ad_group_op1)
 
     # This AdGroup will always fail - campaign ID 0 in resource names is
     # never valid.
-    ad_group_op2 = client.get_type("AdGroupOperation")
+    ad_group_op2: AdGroupOperation = client.get_type("AdGroupOperation")
     ad_group_op2.create.name = f"Broken AdGroup: {uuid.uuid4()}"
     ad_group_op2.create.campaign = invalid_resource_name
     ad_group_operations.append(ad_group_op2)
 
     # This AdGroup will always fail - duplicate ad group names are not allowed.
-    ad_group_op3 = client.get_type("AdGroupOperation")
+    ad_group_op3: AdGroupOperation = client.get_type("AdGroupOperation")
     ad_group_op3.create.name = ad_group_op1.create.name
     ad_group_op3.create.campaign = resource_name
     ad_group_operations.append(ad_group_op3)
 
     # Issue a mutate request, setting partial_failure=True.
-    request = client.get_type("MutateAdGroupsRequest")
+    request: MutateAdGroupsRequest = client.get_type("MutateAdGroupsRequest")
     request.customer_id = customer_id
     request.operations = ad_group_operations
     request.partial_failure = True
@@ -100,7 +122,7 @@ def create_ad_groups(client, customer_id, campaign_id):
 
 
 # [START handle_partial_failure_1]
-def is_partial_failure_error_present(response):
+def is_partial_failure_error_present(response: MutateAdGroupsResponse) -> bool:
     """Checks whether a response message has a partial failure error.
 
     In Python the partial_failure_error attr is always present on a response
@@ -115,14 +137,16 @@ def is_partial_failure_error_present(response):
     Returns: A boolean, whether or not the response message has a partial
         failure error.
     """
-    partial_failure = getattr(response, "partial_failure_error", None)
-    code = getattr(partial_failure, "code", None)
+    partial_failure: Any = getattr(response, "partial_failure_error", None)
+    code: int = int(getattr(partial_failure, "code", 0))  # Default to 0 if None
     return code != 0
     # [END handle_partial_failure_1]
 
 
 # [START handle_partial_failure_2]
-def print_results(client, response):
+def print_results(
+    client: GoogleAdsClient, response: MutateAdGroupsResponse
+) -> None:
     """Prints partial failure errors and success messages from a response.
 
     This function shows how to retrieve partial_failure errors from a response
@@ -163,17 +187,19 @@ def print_results(client, response):
     if is_partial_failure_error_present(response):
         print("Partial failures occurred. Details will be shown below.\n")
         # Prints the details of the partial failure errors.
-        partial_failure = getattr(response, "partial_failure_error", None)
+        partial_failure: Any = getattr(response, "partial_failure_error", None)
         # partial_failure_error.details is a repeated field and iterable
-        error_details = getattr(partial_failure, "details", [])
+        error_details: List[Any] = getattr(partial_failure, "details", [])
 
         for error_detail in error_details:
             # Retrieve an instance of the GoogleAdsFailure class from the client
-            failure_message = client.get_type("GoogleAdsFailure")
+            failure_message: Any = client.get_type("GoogleAdsFailure")
             # Parse the string into a GoogleAdsFailure message instance.
             # To access class-only methods on the message we retrieve its type.
-            GoogleAdsFailure = type(failure_message)
-            failure_object = GoogleAdsFailure.deserialize(error_detail.value)
+            GoogleAdsFailure: Any = type(failure_message)
+            failure_object: Any = GoogleAdsFailure.deserialize(
+                error_detail.value
+            )
 
             for error in failure_object.errors:
                 # Construct and print a string that details which element in
@@ -222,6 +248,8 @@ def print_results(client, response):
 
     # GoogleAdsClient will read the google-ads.yaml configuration file in the
     # home directory if none is specified.
-    googleads_client = GoogleAdsClient.load_from_storage(version="v20")
+    googleads_client: GoogleAdsClient = GoogleAdsClient.load_from_storage(
+        version="v20"
+    )
 
     main(googleads_client, args.customer_id, args.campaign_id)