Created Schedule attach disk sample

Author: Thoughtseize1

compute/client_library/ingredients/snapshots/schedule_attach_disk.py

@@ -0,0 +1,52 @@
+#  Copyright 2024 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
+#  limitations under the License.
+
+# This is an ingredient file. It is not meant to be run directly. Check the samples/snippets
+# folder for complete code samples that are ready to be used.
+# Disabling flake8 for the ingredients file, as it would fail F821 - undefined name check.
+# flake8: noqa
+
+from google.cloud import compute_v1
+
+
+# <INGREDIENT attach_disk_schedule_snapshots>
+def snapshot_schedule_attach(
+    project_id: str, zone: str, region: str, disk_name: str, schedule_name: str
+) -> None:
+    """
+    Attaches a snapshot schedule to a specified disk.
+    Args:
+        project_id (str): The ID of the Google Cloud project.
+        zone (str): The zone where the disk is located.
+        region (str): The region where the snapshot schedule was created
+        disk_name (str): The name of the disk to which the snapshot schedule will be attached.
+        schedule_name (str): The name of the snapshot schedule that you are applying to this disk
+    Returns:
+        None
+    """
+    disks_add_request = compute_v1.DisksAddResourcePoliciesRequest(
+        resource_policies=[f"regions/{region}/resourcePolicies/{schedule_name}"]
+    )
+
+    client = compute_v1.DisksClient()
+    operation = client.add_resource_policies(
+        project=project_id,
+        zone=zone,
+        disk=disk_name,
+        disks_add_resource_policies_request_resource=disks_add_request,
+    )
+    wait_for_extended_operation(operation, "Attaching snapshot schedule to disk")
+
+
+# </INGREDIENT>

compute/client_library/recipes/snapshots/schedule_attach_disk.py

@@ -0,0 +1,23 @@
+#  Copyright 2024 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
+#  limitations under the License.
+# flake8: noqa
+
+# <REGION compute_snapshot_schedule_attach>
+# <IMPORTS/>
+
+# <INGREDIENT wait_for_extended_operation />
+
+# <INGREDIENT attach_disk_schedule_snapshots />
+
+# </REGION compute_snapshot_schedule_attach>

compute/client_library/snippets/snapshots/schedule_attach_disk.py

@@ -0,0 +1,108 @@
+#  Copyright 2024 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
+#  limitations under the License.
+# flake8: noqa
+
+
+# This file is automatically generated. Please do not modify it directly.
+# Find the relevant recipe file in the samples/recipes or samples/ingredients
+# directory and apply your changes there.
+
+
+# [START compute_snapshot_schedule_attach]
+from __future__ import annotations
+
+import sys
+from typing import Any
+
+from google.api_core.extended_operation import ExtendedOperation
+from google.cloud import compute_v1
+
+
+def wait_for_extended_operation(
+    operation: ExtendedOperation, verbose_name: str = "operation", timeout: int = 300
+) -> Any:
+    """
+    Waits for the extended (long-running) operation to complete.
+
+    If the operation is successful, it will return its result.
+    If the operation ends with an error, an exception will be raised.
+    If there were any warnings during the execution of the operation
+    they will be printed to sys.stderr.
+
+    Args:
+        operation: a long-running operation you want to wait on.
+        verbose_name: (optional) a more verbose name of the operation,
+            used only during error and warning reporting.
+        timeout: how long (in seconds) to wait for operation to finish.
+            If None, wait indefinitely.
+
+    Returns:
+        Whatever the operation.result() returns.
+
+    Raises:
+        This method will raise the exception received from `operation.exception()`
+        or RuntimeError if there is no exception set, but there is an `error_code`
+        set for the `operation`.
+
+        In case of an operation taking longer than `timeout` seconds to complete,
+        a `concurrent.futures.TimeoutError` will be raised.
+    """
+    result = operation.result(timeout=timeout)
+
+    if operation.error_code:
+        print(
+            f"Error during {verbose_name}: [Code: {operation.error_code}]: {operation.error_message}",
+            file=sys.stderr,
+            flush=True,
+        )
+        print(f"Operation ID: {operation.name}", file=sys.stderr, flush=True)
+        raise operation.exception() or RuntimeError(operation.error_message)
+
+    if operation.warnings:
+        print(f"Warnings during {verbose_name}:\n", file=sys.stderr, flush=True)
+        for warning in operation.warnings:
+            print(f" - {warning.code}: {warning.message}", file=sys.stderr, flush=True)
+
+    return result
+
+
+def snapshot_schedule_attach(
+    project_id: str, zone: str, region: str, disk_name: str, schedule_name: str
+) -> None:
+    """
+    Attaches a snapshot schedule to a specified disk.
+    Args:
+        project_id (str): The ID of the Google Cloud project.
+        zone (str): The zone where the disk is located.
+        region (str): The region where the snapshot schedule was created
+        disk_name (str): The name of the disk to which the snapshot schedule will be attached.
+        schedule_name (str): The name of the snapshot schedule that you are applying to this disk
+    Returns:
+        None
+    """
+    disks_add_request = compute_v1.DisksAddResourcePoliciesRequest(
+        resource_policies=[f"regions/{region}/resourcePolicies/{schedule_name}"]
+    )
+
+    client = compute_v1.DisksClient()
+    operation = client.add_resource_policies(
+        project=project_id,
+        zone=zone,
+        disk=disk_name,
+        disks_add_resource_policies_request_resource=disks_add_request,
+    )
+    wait_for_extended_operation(operation, "Attaching snapshot schedule to disk")
+
+
+# [END compute_snapshot_schedule_attach]

compute/client_library/snippets/tests/test_snapshots.py

@@ -14,6 +14,8 @@
 import uuid
 
 import google.auth
+from google.cloud import compute_v1
+
 import pytest
 
 from ..disks.create_from_image import create_disk_from_image
@@ -23,6 +25,7 @@
 from ..snapshots.delete import delete_snapshot
 from ..snapshots.get import get_snapshot
 from ..snapshots.list import list_snapshots
+from ..snapshots.schedule_attach_disk import snapshot_schedule_attach
 from ..snapshots.schedule_create import snapshot_schedule_create
 from ..snapshots.schedule_delete import snapshot_schedule_delete
 from ..snapshots.schedule_get import snapshot_schedule_get
@@ -50,6 +53,20 @@ def test_disk():
     delete_disk(PROJECT, ZONE, test_disk_name)
 
 
+@pytest.fixture
+def test_schedule_snapshot():
+    test_schedule_snapshot_name = "test-snapshot-" + uuid.uuid4().hex[:5]
+    schedule_snapshot = snapshot_schedule_create(
+        PROJECT,
+        REGION,
+        test_schedule_snapshot_name,
+        "test description",
+        {"env": "dev", "media": "images"},
+    )
+    yield schedule_snapshot
+    snapshot_schedule_delete(PROJECT, REGION, test_schedule_snapshot_name)
+
+
 def test_snapshot_create_delete(test_disk):
     snapshot_name = "test-snapshot-" + uuid.uuid4().hex[:10]
     snapshot = create_snapshot(PROJECT, test_disk.name, snapshot_name, zone=ZONE)
@@ -93,3 +110,12 @@ def test_create_get_list_delete_schedule_snapshot():
     finally:
         snapshot_schedule_delete(PROJECT, REGION, test_snapshot_name)
         assert len(list(snapshot_schedule_list(PROJECT, REGION))) == 0
+
+
+def test_attach_disk_to_snapshot(test_schedule_snapshot, test_disk):
+    assert not test_disk.resource_policies
+    snapshot_schedule_attach(
+        PROJECT, ZONE, REGION, test_disk.name, test_schedule_snapshot.name
+    )
+    disk = compute_v1.DisksClient().get(project=PROJECT, zone=ZONE, disk=test_disk.name)
+    assert test_schedule_snapshot.name in disk.resource_policies[0]