chore(secretmanager): Add regional samples for delayed destory (#13317)

Author: archana-9430

secretmanager/snippets/regional_samples/create_regional_secret_with_delayed_destroy.py

@@ -0,0 +1,94 @@
+#!/usr/bin/env python
+
+# Copyright 2025 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
+"""
+Command line application and sample code for creating a new secret with
+delayed destroy.
+"""
+
+import argparse
+
+# [START secretmanager_create_regional_secret_with_delayed_destroy]
+
+# Import the Secret Manager client library.
+from google.cloud import secretmanager_v1
+from google.protobuf.duration_pb2 import Duration
+
+
+def create_regional_secret_with_delayed_destroy(
+    project_id: str,
+    location_id: str,
+    secret_id: str,
+    version_destroy_ttl: int,
+) -> secretmanager_v1.Secret:
+    """
+    Create a new secret with the given name and version_destroy_ttl. A secret is a logical wrapper
+    around a collection of secret versions. Secret versions hold the actual
+    secret material.
+
+    Args:
+      project_id: Parent project id
+      location_id: Location of the secret
+      secret_id: ID of the secret or fully qualified identifier for the secret
+      version_destroy_ttl: Secret Version TTL after destruction request
+
+    Returns:
+      Regional secret with delayed destroy
+      """
+
+    # Endpoint to call the regional secret manager sever
+    api_endpoint = f"secretmanager.{location_id}.rep.googleapis.com"
+
+    # Create the Secret Manager client.
+    client = secretmanager_v1.SecretManagerServiceClient(
+        client_options={"api_endpoint": api_endpoint},
+    )
+
+    # Build the resource name of the parent project.
+    parent = f"projects/{project_id}/locations/{location_id}"
+
+    # Create the secret.
+    response = client.create_secret(
+        request={
+            "parent": parent,
+            "secret_id": secret_id,
+            "secret": {"version_destroy_ttl": Duration(seconds=version_destroy_ttl)},
+        }
+    )
+
+    # Print the new secret name.
+    print(f"Created secret: {response.name}")
+
+    return response
+
+# [END secretmanager_create_regional_secret_with_delayed_destroy]
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(
+        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
+    )
+    parser.add_argument("project_id", help="id of the GCP project")
+    parser.add_argument(
+        "location_id", help="id of the location where secret is to be created"
+    )
+    parser.add_argument("secret_id", help="id of the secret to create")
+    parser.add_argument(
+        "version_destroy_ttl", help="version_destroy_ttl you want to add (in seconds)"
+    )
+    args = parser.parse_args()
+
+    create_regional_secret_with_delayed_destroy(
+        args.project_id, args.location_id, args.secret_id, int(args.version_destroy_ttl)
+    )

secretmanager/snippets/regional_samples/disable_regional_secret_delayed_destroy.py

@@ -0,0 +1,78 @@
+#!/usr/bin/env python
+
+# Copyright 2025 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
+
+import argparse
+
+# [START secretmanager_disable_regional_secret_delayed_destroy]
+
+# Import the Secret Manager client library.
+from google.cloud import secretmanager_v1
+
+
+def disable_regional_secret_delayed_destroy(
+    project_id: str, location_id: str, secret_id: str
+) -> secretmanager_v1.Secret:
+    """
+    Disable delayed destroy on an existing secret with a version destroy ttl.
+
+    Args:
+      project_id: Parent project id
+      location_id: Location of the secret
+      secret_id: ID of the secret or fully qualified identifier for the secret
+
+    Returns:
+      Regional secret with disabled delayed destroy
+    """
+
+    # Endpoint to call the regional secret manager sever
+    api_endpoint = f"secretmanager.{location_id}.rep.googleapis.com"
+
+    # Create the Secret Manager client.
+    client = secretmanager_v1.SecretManagerServiceClient(
+        client_options={"api_endpoint": api_endpoint},
+    )
+
+    # Build the resource name.
+    name = f"projects/{project_id}/locations/{location_id}/secrets/{secret_id}"
+
+    # Disable delayed destroy on secret
+    secret = {"name": name}
+    update_mask = {"paths": ["version_destroy_ttl"]}
+    response = client.update_secret(
+        request={"secret": secret, "update_mask": update_mask}
+    )
+
+    # Print the new secret name.
+    print(f"Disabled delayed destroy on secret: {response.name}")
+
+    return response
+
+# [END secretmanager_disable_regional_secret_delayed_destroy]
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(
+        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
+    )
+    parser.add_argument("project_id", help="id of the GCP project")
+    parser.add_argument(
+        "location_id", help="id of the location where secret is to be created"
+    )
+    parser.add_argument("secret_id", help="id of the secret to act on")
+    args = parser.parse_args()
+
+    disable_regional_secret_delayed_destroy(
+        args.project_id, args.location_id, args.secret_id
+    )

secretmanager/snippets/regional_samples/snippets_test.py

@@ -11,25 +11,29 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 
+from datetime import timedelta
 import os
 import time
 from typing import Iterator, Tuple, Union
 import uuid
 
 from google.api_core import exceptions, retry
 from google.cloud import secretmanager_v1
+from google.protobuf.duration_pb2 import Duration
 import pytest
 
 from regional_samples import access_regional_secret_version
 from regional_samples import add_regional_secret_version
 from regional_samples import create_regional_secret
 from regional_samples import create_regional_secret_with_annotations
+from regional_samples import create_regional_secret_with_delayed_destroy
 from regional_samples import create_regional_secret_with_labels
 from regional_samples import delete_regional_secret
 from regional_samples import delete_regional_secret_label
 from regional_samples import delete_regional_secret_with_etag
 from regional_samples import destroy_regional_secret_version
 from regional_samples import destroy_regional_secret_version_with_etag
+from regional_samples import disable_regional_secret_delayed_destroy
 from regional_samples import disable_regional_secret_version
 from regional_samples import disable_regional_secret_version_with_etag
 from regional_samples import edit_regional_secret_annotations
@@ -46,6 +50,7 @@
 from regional_samples import list_regional_secrets_with_filter
 from regional_samples import regional_quickstart
 from regional_samples import update_regional_secret
+from regional_samples import update_regional_secret_with_delayed_destroy
 from regional_samples import update_regional_secret_with_etag
 from regional_samples import view_regional_secret_annotations
 from regional_samples import view_regional_secret_labels
@@ -99,6 +104,11 @@ def annotation_value() -> str:
     return "annotationvalue"
 
 
+@pytest.fixture()
+def version_destroy_ttl() -> int:
+    return 604800  # 7 days in seconds
+
+
 @retry.Retry()
 def retry_client_create_regional_secret(
     regional_client: secretmanager_v1.SecretManagerServiceClient,
@@ -201,6 +211,33 @@ def regional_secret(
     yield secret_id, regional_secret.etag
 
 
+@pytest.fixture()
+def regional_secret_with_delayed_destroy(
+    regional_client: secretmanager_v1.SecretManagerServiceClient,
+    project_id: str,
+    location_id: str,
+    secret_id: str,
+    version_destroy_ttl: int,
+) -> Iterator[str]:
+    print("creating secret with given secret id.")
+
+    parent = f"projects/{project_id}/locations/{location_id}"
+    time.sleep(5)
+    retry_client_create_regional_secret(
+        regional_client,
+        request={
+            "parent": parent,
+            "secret_id": secret_id,
+            "secret": {
+                "version_destroy_ttl": Duration(seconds=version_destroy_ttl),
+            },
+        },
+    )
+    print("debug")
+
+    yield secret_id
+
+
 def test_regional_quickstart(project_id: str, location_id: str, secret_id: str) -> None:
     regional_quickstart.regional_quickstart(project_id, location_id, secret_id)
 
@@ -259,6 +296,18 @@ def test_create_regional_secret_with_annotations(
     assert secret_id in secret.name
 
 
+def test_create_regional_secret_with_delayed_destroy(
+    regional_client: secretmanager_v1.SecretManagerServiceClient,
+    project_id: str,
+    location_id: str,
+    secret_id: str,
+    version_destroy_ttl: int,
+) -> None:
+    secret = create_regional_secret_with_delayed_destroy.create_regional_secret_with_delayed_destroy(project_id, location_id, secret_id, version_destroy_ttl)
+    assert secret_id in secret.name
+    assert timedelta(seconds=version_destroy_ttl) == secret.version_destroy_ttl
+
+
 def test_create_regional_secret_with_label(
     regional_client: secretmanager_v1.SecretManagerServiceClient,
     project_id: str,
@@ -336,6 +385,17 @@ def test_destroy_regional_secret_version_with_etag(
     assert version.destroy_time
 
 
+def test_disable_regional_secret_delayed_destroy(
+    regional_client: secretmanager_v1.SecretManagerServiceClient,
+    regional_secret_with_delayed_destroy: str,
+    project_id: str,
+    location_id: str,
+) -> None:
+    secret_id = regional_secret_with_delayed_destroy
+    updated_secret = disable_regional_secret_delayed_destroy.disable_regional_secret_delayed_destroy(project_id, location_id, secret_id)
+    assert updated_secret.version_destroy_ttl == timedelta(0)
+
+
 def test_enable_disable_regional_secret_version(
     regional_client: secretmanager_v1.SecretManagerServiceClient,
     regional_secret_version: Tuple[str, str, str],
@@ -612,6 +672,18 @@ def test_update_regional_secret(
     assert updated_regional_secret.labels["secretmanager"] == "rocks"
 
 
+def test_update_regional_secret_with_delayed_destroy(
+    regional_secret_with_delayed_destroy: str,
+    project_id: str,
+    location_id: str,
+    version_destroy_ttl: int
+) -> None:
+    secret_id = regional_secret_with_delayed_destroy
+    updated_version_delayed_destroy = 118400
+    updated_secret = update_regional_secret_with_delayed_destroy.update_regional_secret_with_delayed_destroy(project_id, location_id, secret_id, updated_version_delayed_destroy)
+    assert updated_secret.version_destroy_ttl == timedelta(seconds=updated_version_delayed_destroy)
+
+
 def test_view_regional_secret_labels(
     capsys: pytest.LogCaptureFixture,
     project_id: str,