docs(samples): Update Topic with Kinesis Ingestion Settings

mukund-ananthu · mukund-ananthu · commit e01d3f163a36 · 2024-03-14T21:04:02.000Z
Add code to demonstrate updating a topic with Kinesis Ingestion Settings IngestionDataSourceSettings were added as an attribute to Topic [here](https://github.com/googleapis/googleapis/blob/65277ddce9caa1cfd1a0eb7ab67980fc73d20b50/google/pubsub/v1/pubsub.proto#L316)
diff --git a/samples/snippets/publisher.py b/samples/snippets/publisher.py
@@ -105,6 +105,58 @@ def create_topic_kinesis_ingestion(
     # [END pubsub_create_topic_kinesis_ingestion]
 
 
+def update_topic_kinesis_ingestion(
+    project_id: str,
+    topic_id: str,
+    stream_arn: str,
+    consumer_arn: str,
+    aws_role_arn: str,
+    gcp_service_account: str,
+) -> None:
+    """Update Pub/Sub topic with AWS Kinesis Ingestion Settings."""
+    # [START pubsub_quickstart_update_topic_kinesis_ingestion]
+    # [START pubsub_update_topic_kinesis_ingestion]
+    from google.cloud import pubsub_v1
+    from google.pubsub_v1.types import Topic
+    from google.pubsub_v1.types import IngestionDataSourceSettings
+    from google.pubsub_v1.types import UpdateTopicRequest
+    from google.protobuf import field_mask_pb2
+
+    # TODO(developer)
+    # project_id = "your-project-id"
+    # topic_id = "your-topic-id"
+    # stream_arn = "your-stream-arn"
+    # consumer_arn = "your-consumer-arn"
+    # aws_role_arn = "your-aws-role-arn"
+    # gcp_service_account = "your-gcp-service-account"
+
+    publisher = pubsub_v1.PublisherClient()
+    topic_path = publisher.topic_path(project_id, topic_id)
+
+    update_request = UpdateTopicRequest(
+        topic=Topic(
+            name=topic_path,
+            ingestion_data_source_settings=IngestionDataSourceSettings(
+                aws_kinesis=IngestionDataSourceSettings.AwsKinesis(
+                    stream_arn=stream_arn,
+                    consumer_arn=consumer_arn,
+                    aws_role_arn=aws_role_arn,
+                    gcp_service_account=gcp_service_account,
+                )
+            ),
+        ),
+        update_mask=field_mask_pb2.FieldMask(
+            paths=["ingestion_data_source_settings"]),
+    )
+
+    topic = publisher.update_topic(request=update_request)
+    print(f"Updated topic: {topic.name} with AWS Kinesis Ingestion Settings")
+
+
+# [END pubsub_quickstart_update_topic_kinesis_ingestion]
+# [END pubsub_update_topic_kinesis_ingestion]
+
+
 def delete_topic(project_id: str, topic_id: str) -> None:
     """Deletes an existing Pub/Sub topic."""
     # [START pubsub_delete_topic]
@@ -483,6 +535,15 @@ def detach_subscription(project_id: str, subscription_id: str) -> None:
     create_topic_kinesis_ingestion_parser.add_argument("consumer_arn")
     create_topic_kinesis_ingestion_parser.add_argument("aws_role_arn")
     create_topic_kinesis_ingestion_parser.add_argument("gcp_service_account")
+    
+    update_topic_kinesis_ingestion_parser = subparsers.add_parser(
+        "update_kinesis_ingestion", help=update_topic_kinesis_ingestion.__doc__
+    )
+    update_topic_kinesis_ingestion_parser.add_argument("topic_id")
+    update_topic_kinesis_ingestion_parser.add_argument("stream_arn")
+    update_topic_kinesis_ingestion_parser.add_argument("consumer_arn")
+    update_topic_kinesis_ingestion_parser.add_argument("aws_role_arn")
+    update_topic_kinesis_ingestion_parser.add_argument("gcp_service_account")
 
     delete_parser = subparsers.add_parser("delete", help=delete_topic.__doc__)
     delete_parser.add_argument("topic_id")
@@ -553,6 +614,15 @@ def detach_subscription(project_id: str, subscription_id: str) -> None:
             args.aws_role_arn,
             args.gcp_service_account,
         )
+    elif args.command == "update_kinesis_ingestion":
+        update_topic_kinesis_ingestion(
+            args.project_id,
+            args.topic_id,
+            args.stream_arn,
+            args.consumer_arn,
+            args.aws_role_arn,
+            args.gcp_service_account,
+        )
     elif args.command == "delete":
         delete_topic(args.project_id, args.topic_id)
     elif args.command == "publish":
diff --git a/samples/snippets/publisher_test.py b/samples/snippets/publisher_test.py
@@ -123,6 +123,9 @@ def test_create(
     out, _ = capsys.readouterr()
     assert f"Created topic: {topic_path}" in out
 
+    # Clean up resource created for the test.
+    publisher_client.delete_topic(request={"topic": topic_path})
+
 
 def test_create_kinesis_ingestion(
     publisher_client: pubsub_v1.PublisherClient, capsys: CaptureFixture[str]
@@ -155,6 +158,48 @@ def test_create_kinesis_ingestion(
     out, _ = capsys.readouterr()
     assert f"Created topic: {topic_path} with AWS Kinesis Ingestion Settings" in out
 
+    # Clean up resource created for the test.
+    publisher_client.delete_topic(request={"topic": topic_path})
+    
+def test_update_kinesis_ingestion(
+    publisher_client: pubsub_v1.PublisherClient, capsys: CaptureFixture[str]
+) -> None:
+    # The scope of `topic_path` is limited to this function.
+    topic_path = publisher_client.topic_path(PROJECT_ID, TOPIC_ID)
+
+    # Outside of automated CI tests, these values must be of actual AWS resources for the test to pass.
+    stream_arn = "arn:aws:kinesis:us-west-2:111111111111:stream/fake-stream-name"
+    consumer_arn = "arn:aws:kinesis:us-west-2:111111111111:stream/fake-stream-name/consumer/consumer-1:1111111111"
+    aws_role_arn = "arn:aws:iam::111111111111:role/fake-role-name"
+    gcp_service_account = (
+        "fake-service-account@fake-gcp-project.iam.gserviceaccount.com"
+    )
+
+    try:
+        publisher_client.delete_topic(request={"topic": topic_path})
+    except NotFound:
+        pass
+
+    publisher.create_topic(PROJECT_ID, TOPIC_ID)
+
+    out, _ = capsys.readouterr()
+    assert f"Created topic: {topic_path}" in out
+    
+    publisher.update_topic_kinesis_ingestion(
+        PROJECT_ID,
+        TOPIC_ID,
+        stream_arn,
+        consumer_arn,
+        aws_role_arn,
+        gcp_service_account,
+    )
+    
+    out, _ = capsys.readouterr()
+    assert f"Updated topic: {topic_path} with AWS Kinesis Ingestion Settings" in out
+    
+    # Clean up resource created for the test.
+    publisher_client.delete_topic(request={"topic": topic_path})
+
 
 def test_list(topic_path: str, capsys: CaptureFixture[str]) -> None:
     publisher.list_topics(PROJECT_ID)
