googleapis · kamalaboulhosn · Apr 28, 2025 · Jan 24, 2023 · Jan 24, 2023 · Feb 24, 2023
@@ -326,6 +326,39 @@ def create_topic_with_confluent_cloud_ingestion(
     # [END pubsub_create_topic_with_confluent_cloud_ingestion]
 
 
+def create_topic_with_smt(
+    project_id: str,
+    topic_id: str,
+) -> None:
+    """Create a new Pub/Sub topic with a UDF SMT."""
+    # [START pubsub_create_topic_with_smt]
+    from google.cloud import pubsub_v1
+    from google.pubsub_v1.types import JavaScriptUDF, MessageTransform, Topic
+
+    # TODO(developer)
+    # project_id = "your-project-id"
+    # topic_id = "your-topic-id"
+
+    code = """function redactSSN(message, metadata) {
+                const data = JSON.parse(message.data);
+                delete data['ssn'];
+                message.data = JSON.stringify(data);
+                return message;
+                }"""
+    udf = JavaScriptUDF(code=code, function_name="redactSSN")
+    transforms = [MessageTransform(javascript_udf=udf)]
+
+    publisher = pubsub_v1.PublisherClient()
+    topic_path = publisher.topic_path(project_id, topic_id)
+
+    request = Topic(name=topic_path, message_transforms=transforms)
+
+    topic = publisher.create_topic(request=request)
+
+    print(f"Created topic: {topic.name} with SMT")
+    # [END pubsub_create_topic_with_smt]
+
+
 def update_topic_type(
     project_id: str,
     topic_id: str,
@@ -888,6 +921,11 @@ def detach_subscription(project_id: str, subscription_id: str) -> None:
         "gcp_service_account"
     )
 
+    create_parser = subparsers.add_parser(
+        "create_smt", help=create_topic_with_smt.__doc__
+    )
+    create_parser.add_argument("topic_id")
+
     update_topic_type_parser = subparsers.add_parser(
         "update_kinesis_ingestion", help=update_topic_type.__doc__
     )
@@ -1007,6 +1045,11 @@ def detach_subscription(project_id: str, subscription_id: str) -> None:
             args.identity_pool_id,
             args.gcp_service_account,
         )
+    elif args.command == "create_smt":
+        create_topic_with_smt(
+            args.project_id,
+            args.topic_id,
+        )
     elif args.command == "update_kinesis_ingestion":
         update_topic_type(
             args.project_id,

@@ -313,6 +313,26 @@ def test_create_topic_with_confluent_cloud_ingestion(
     publisher_client.delete_topic(request={"topic": topic_path})
 
 
+def test_create_with_smt(
+    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)
+
+    try:
+        publisher_client.delete_topic(request={"topic": topic_path})
+    except NotFound:
+        pass
+
+    publisher.create_topic_with_smt(PROJECT_ID, TOPIC_ID)
+
+    out, _ = capsys.readouterr()
+    assert f"Created topic: {topic_path} with SMT" in out
+
+    # Clean up resource created for the test.
+    publisher_client.delete_topic(request={"topic": topic_path})
+
+
 def test_update_topic_type(
     publisher_client: pubsub_v1.PublisherClient, capsys: CaptureFixture[str]
 ) -> None:

@@ -578,6 +578,45 @@ def create_cloudstorage_subscription(
     # [END pubsub_create_cloud_storage_subscription]
 
 
+def create_subscription_with_smt(
+    project_id: str, topic_id: str, subscription_id: str
+) -> None:
+    """Create a subscription with a UDF SMT."""
+    # [START pubsub_create_subscription_with_smt]
+    from google.cloud import pubsub_v1
+    from google.pubsub_v1.types import JavaScriptUDF, MessageTransform
+
+    # TODO(developer): Choose an existing topic.
+    # project_id = "your-project-id"
+    # topic_id = "your-topic-id"
+    # subscription_id = "your-subscription-id"
+
+    publisher = pubsub_v1.PublisherClient()
+    subscriber = pubsub_v1.SubscriberClient()
+    topic_path = publisher.topic_path(project_id, topic_id)
+    subscription_path = subscriber.subscription_path(project_id, subscription_id)
+
+    code = """function redactSSN(message, metadata) {
+                const data = JSON.parse(message.data);
+                delete data['ssn'];
+                message.data = JSON.stringify(data);
+                return message;
+                }"""
+    udf = JavaScriptUDF(code=code, function_name="redactSSN")
+    transforms = [MessageTransform(javascript_udf=udf)]
+
+    with subscriber:
+        subscription = subscriber.create_subscription(
+            request={
+                "name": subscription_path,
+                "topic": topic_path,
+                "message_transforms": transforms,
+            }
+        )
+        print(f"Created subscription with SMT: {subscription}")
+    # [END pubsub_create_subscription_with_smt]
+
+
 def delete_subscription(project_id: str, subscription_id: str) -> None:
     """Deletes an existing Pub/Sub topic."""
     # [START pubsub_delete_subscription]
@@ -1310,6 +1349,12 @@ def callback(message: pubsub_v1.subscriber.message.Message) -> None:
     create_cloudstorage_subscription_parser.add_argument("subscription_id")
     create_cloudstorage_subscription_parser.add_argument("bucket")
 
+    create_subscription_with_smt_parser = subparsers.add_parser(
+        "create-with-smt", help=create_subscription_with_smt.__doc__
+    )
+    create_subscription_with_smt_parser.add_argument("topic_id")
+    create_subscription_with_smt_parser.add_argument("subscription_id")
+
     delete_parser = subparsers.add_parser("delete", help=delete_subscription.__doc__)
     delete_parser.add_argument("subscription_id")
 
@@ -1471,6 +1516,10 @@ def callback(message: pubsub_v1.subscriber.message.Message) -> None:
         create_cloudstorage_subscription(
             args.project_id, args.topic_id, args.subscription_id, args.bucket
         )
+    elif args.command == "create-with-smt":
+        create_subscription_with_smt(
+            args.project_id, args.topic_id, args.subscription_id
+        )
 
     elif args.command == "delete":
         delete_subscription(args.project_id, args.subscription_id)

@@ -579,6 +579,37 @@ def test_create_push_subscription(
     subscriber_client.delete_subscription(request={"subscription": subscription_path})
 
 
+def test_create_subscription_with_smt(
+    subscriber_client: pubsub_v1.SubscriberClient,
+    topic: str,
+    capsys: CaptureFixture[str],
+) -> None:
+    subscription_for_create_name = (
+        f"subscription-test-subscription-for-create-with-smt-{PY_VERSION}-{UUID}"
+    )
+
+    subscription_path = subscriber_client.subscription_path(
+        PROJECT_ID, subscription_for_create_name
+    )
+
+    try:
+        subscriber_client.delete_subscription(
+            request={"subscription": subscription_path}
+        )
+    except NotFound:
+        pass
+
+    subscriber.create_subscription_with_smt(
+        PROJECT_ID, TOPIC, subscription_for_create_name
+    )
+
+    out, _ = capsys.readouterr()
+    assert f"{subscription_for_create_name}" in out
+
+    # Clean up.
+    subscriber_client.delete_subscription(request={"subscription": subscription_path})
+
+
 def test_update_push_subscription(
     subscriber_client: pubsub_v1.SubscriberClient,
     topic: str,