docs(secretmanager): add sample code for receiving a Pub/Sub message

samples/snippets/consume_event_notification.py

@@ -0,0 +1,35 @@
+#!/usr/bin/env python
+
+# Copyright 2021 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+"""
+sample code for consuming an event notification in a cloud function.
+"""
+
+import base64
+
+
+# [START secretmanager_consume_event_notification]
+def consume_event_notification(event, unused_context):
+    """
+    consume_event_notification demonstrates how to consume and process a
+    Pub/Sub notification from Secret Manager.
+    Args:
+          event (dict): Event payload.
+          unused_context (google.cloud.functions.Context): Metadata for the event.
+    """
+    event_type = event['attributes']['eventType']
+    secret_id = event['attributes']['secretId']
+    secret_metadata = base64.b64decode(event['data']).decode('utf-8')
+    return f'Received {event_type} for {secret_id}. New metadata: {secret_metadata}'
+# [END secretmanager_consume_event_notification]

samples/snippets/snippets_test.py

@@ -11,6 +11,7 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 
+import base64
 import os
 import uuid
 
@@ -20,6 +21,7 @@
 
 from access_secret_version import access_secret_version
 from add_secret_version import add_secret_version
+from consume_event_notification import consume_event_notification
 from create_secret import create_secret
 from delete_secret import delete_secret
 from delete_secret_with_etag import delete_secret_with_etag
@@ -99,6 +101,17 @@ def secret_version(client, secret):
 another_secret_version = secret_version
 
 
+@pytest.fixture()
+def pubsub_message():
+    message = "hello!"
+    message_bytes = message.encode()
+    base64_bytes = base64.b64encode(message_bytes)
+    return {
+        "attributes": {"eventType": "SECRET_UPDATE", "secretId": "projects/p/secrets/s"},
+        "data": base64_bytes
+    }
+
+
 def test_quickstart(project_id):
     secret_id = "python-secret-{}".format(uuid.uuid4())
     quickstart(project_id, secret_id)
@@ -224,6 +237,11 @@ def test_update_secret(secret):
     assert secret.labels["secretmanager"] == "rocks"
 
 
+def test_consume_event_notification(pubsub_message):
+    got = consume_event_notification(pubsub_message, None)
+    assert got == "Received SECRET_UPDATE for projects/p/secrets/s. New metadata: hello!"
+
+
 def test_update_secret_with_etag(secret):
     project_id, secret_id, etag = secret
     secret = update_secret_with_etag(project_id, secret_id, etag)