docs(samples): Improving snapshot samples (#276)

Improving the code snippets for disk snapshots, so they can be used on this page: [Create and manage disk snapshots
](https://cloud.google.com/compute/docs/disks/create-snapshots)

Author: m-strzelczyk

compute/compute/ingredients/operations/handle_extended_operation.py

@@ -55,14 +55,15 @@ def wait_for_extended_operation(
     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)
-        print(f"Operation ID: {operation.name}")
+        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)
+        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)
+            print(f" - {warning.code}: {warning.message}", file=sys.stderr, flush=True)
 
     return result
 # </INGREDIENT>

compute/compute/ingredients/snapshots/create.py

@@ -16,30 +16,60 @@
 # 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 typing import Optional
 
 from google.cloud import compute_v1
 
 
 # <INGREDIENT create_snapshot>
-def create_snapshot(project_id: str, zone: str, disk_name: str, snapshot_name: str) -> compute_v1.Snapshot:
+def create_snapshot(project_id: str, disk_name: str, snapshot_name: str, *,
+                    zone: Optional[str] = None, region: Optional[str] = None,
+                    location: Optional[str] = None, disk_project_id: Optional[str] = None) -> compute_v1.Snapshot:
     """
     Create a snapshot of a disk.
 
+    You need to pass `zone` or `region` parameter relevant to the disk you want to
+    snapshot, but not both. Pass `zone` parameter for zonal disks and `region` for
+    regional disks.
+
     Args:
-        project_id: project ID or project number of the Cloud project you want to use.
-        zone: name of the zone in which is the disk you want to snapshot.
+        project_id: project ID or project number of the Cloud project you want
+            to use to store the snapshot.
         disk_name: name of the disk you want to snapshot.
         snapshot_name: name of the snapshot to be created.
+        zone: name of the zone in which is the disk you want to snapshot (for zonal disks).
+        region: name of the region in which is the disk you want to snapshot (for regional disks).
+        location: The Cloud Storage multi-region or the Cloud Storage region where you
+            want to store your snapshot.
+            You can specify only one storage location. Available locations:
+            https://cloud.google.com/storage/docs/locations#available-locations
+        disk_project_id: project ID or project number of the Cloud project that
+            hosts the disk you want to snapshot. If not provided, will look for
+            the disk in the `project_id` project.
 
     Returns:
         The new snapshot instance.
     """
-    disk_client = compute_v1.DisksClient()
-    disk = disk_client.get(project=project_id, zone=zone, disk=disk_name)
+    if zone is None and region is None:
+        raise RuntimeError("You need to specify `zone` or `region` for this function to work.")
+    if zone is not None and region is not None:
+        raise RuntimeError("You can't set both `zone` and `region` parameters.")
+
+    if disk_project_id is None:
+        disk_project_id = project_id
+
+    if zone is not None:
+        disk_client = compute_v1.DisksClient()
+        disk = disk_client.get(project=disk_project_id, zone=zone, disk=disk_name)
+    else:
+        regio_disk_client = compute_v1.RegionDisksClient()
+        disk = regio_disk_client.get(project=disk_project_id, region=region, disk=disk_name)
+
     snapshot = compute_v1.Snapshot()
     snapshot.source_disk = disk.self_link
     snapshot.name = snapshot_name
+    if location:
+        snapshot.storage_locations = [location]
 
     snapshot_client = compute_v1.SnapshotsClient()
     operation = snapshot_client.insert(project=project_id, snapshot_resource=snapshot)

compute/compute/ingredients/snapshots/get.py

@@ -0,0 +1,42 @@
+#  Copyright 2022 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 typing import Iterable
+
+from google.cloud import compute_v1
+
+
+# <INGREDIENT get_snapshot>
+def get_snapshot(project_id: str, snapshot_name: str) -> compute_v1.Snapshot:
+    """
+    Get information about a Snapshot.
+
+    Args:
+        project_id: project ID or project number of the Cloud project you want to use.
+        snapshot_name: the name of the snapshot you want to look up.
+
+    Returns:
+        A Snapshot object.
+    """
+
+    snapshot_client = compute_v1.SnapshotsClient()
+
+    return snapshot_client.get(project=project_id, snapshot=snapshot_name)
+# </INGREDIENT>
+
+

compute/compute/ingredients/snapshots/list.py

@@ -22,13 +22,13 @@
 
 
 # <INGREDIENT list_snapshots>
-def list_snapshots(project_id: str, filter_: str = "") -> Iterable[compute_v1.Snapshot]:
+def list_snapshots(project_id: str, filter: str = "") -> Iterable[compute_v1.Snapshot]:
     """
     List snapshots from a project.
 
     Args:
         project_id: project ID or project number of the Cloud project you want to use.
-        filter_: filter to be applied when listing snapshots. Learn more about filters here:
+        filter: filter to be applied when listing snapshots. Learn more about filters here:
             https://cloud.google.com/python/docs/reference/compute/latest/google.cloud.compute_v1.types.ListSnapshotsRequest
 
     Returns:
@@ -38,7 +38,7 @@ def list_snapshots(project_id: str, filter_: str = "") -> Iterable[compute_v1.Sn
     snapshot_client = compute_v1.SnapshotsClient()
     request = compute_v1.ListSnapshotsRequest()
     request.project = project_id
-    request.filter = filter_
+    request.filter = filter
 
     return snapshot_client.list(request)
 # </INGREDIENT>

compute/compute/recipes/snapshots/delete_by_filter.py

@@ -0,0 +1,37 @@
+#  Copyright 2022 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_delete_by_filter>
+# <IMPORTS/>
+
+# <INGREDIENT wait_for_extended_operation />
+
+# <INGREDIENT delete_snapshot />
+
+# <INGREDIENT list_snapshots />
+
+def delete_snapshots_by_filter(project_id: str, filter: str):
+    """
+    Deletes all snapshots in project that meet the filter criteria.
+
+    Args:
+        project_id: project ID or project number of the Cloud project you want to use.
+        filter: filter to be applied when looking for snapshots for deletion.
+    """
+    for snapshot in list_snapshots(project_id, filter):
+        delete_snapshot(project_id, snapshot.name)
+
+# </REGION compute_snapshot_delete_by_filter>

compute/compute/recipes/snapshots/get.py

@@ -0,0 +1,21 @@
+#  Copyright 2022 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_get>
+# <IMPORTS/>
+
+# <INGREDIENT get_snapshot />
+
+# </REGION compute_snapshot_get>

compute/compute/snippets/disks/autodelete_change.py

@@ -61,14 +61,15 @@ def wait_for_extended_operation(
         print(
             f"Error during {verbose_name}: [Code: {operation.error_code}]: {operation.error_message}",
             file=sys.stderr,
+            flush=True,
         )
-        print(f"Operation ID: {operation.name}")
+        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)
+        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)
+            print(f" - {warning.code}: {warning.message}", file=sys.stderr, flush=True)
 
     return result
 

compute/compute/snippets/disks/create_empty_disk.py

@@ -61,14 +61,15 @@ def wait_for_extended_operation(
         print(
             f"Error during {verbose_name}: [Code: {operation.error_code}]: {operation.error_message}",
             file=sys.stderr,
+            flush=True,
         )
-        print(f"Operation ID: {operation.name}")
+        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)
+        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)
+            print(f" - {warning.code}: {warning.message}", file=sys.stderr, flush=True)
 
     return result
 

compute/compute/snippets/disks/create_from_image.py

@@ -61,14 +61,15 @@ def wait_for_extended_operation(
         print(
             f"Error during {verbose_name}: [Code: {operation.error_code}]: {operation.error_message}",
             file=sys.stderr,
+            flush=True,
         )
-        print(f"Operation ID: {operation.name}")
+        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)
+        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)
+            print(f" - {warning.code}: {warning.message}", file=sys.stderr, flush=True)
 
     return result
 

compute/compute/snippets/disks/create_from_snapshot.py

@@ -61,14 +61,15 @@ def wait_for_extended_operation(
         print(
             f"Error during {verbose_name}: [Code: {operation.error_code}]: {operation.error_message}",
             file=sys.stderr,
+            flush=True,
         )
-        print(f"Operation ID: {operation.name}")
+        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)
+        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)
+            print(f" - {warning.code}: {warning.message}", file=sys.stderr, flush=True)
 
     return result
 

compute/compute/snippets/disks/delete.py

@@ -61,14 +61,15 @@ def wait_for_extended_operation(
         print(
             f"Error during {verbose_name}: [Code: {operation.error_code}]: {operation.error_message}",
             file=sys.stderr,
+            flush=True,
         )
-        print(f"Operation ID: {operation.name}")
+        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)
+        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)
+            print(f" - {warning.code}: {warning.message}", file=sys.stderr, flush=True)
 
     return result
 

compute/compute/snippets/firewall/create.py

@@ -61,14 +61,15 @@ def wait_for_extended_operation(
         print(
             f"Error during {verbose_name}: [Code: {operation.error_code}]: {operation.error_message}",
             file=sys.stderr,
+            flush=True,
         )
-        print(f"Operation ID: {operation.name}")
+        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)
+        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)
+            print(f" - {warning.code}: {warning.message}", file=sys.stderr, flush=True)
 
     return result
 

compute/compute/snippets/firewall/delete.py

@@ -61,14 +61,15 @@ def wait_for_extended_operation(
         print(
             f"Error during {verbose_name}: [Code: {operation.error_code}]: {operation.error_message}",
             file=sys.stderr,
+            flush=True,
         )
-        print(f"Operation ID: {operation.name}")
+        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)
+        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)
+            print(f" - {warning.code}: {warning.message}", file=sys.stderr, flush=True)
 
     return result
 

compute/compute/snippets/firewall/patch.py

@@ -61,14 +61,15 @@ def wait_for_extended_operation(
         print(
             f"Error during {verbose_name}: [Code: {operation.error_code}]: {operation.error_message}",
             file=sys.stderr,
+            flush=True,
         )
-        print(f"Operation ID: {operation.name}")
+        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)
+        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)
+            print(f" - {warning.code}: {warning.message}", file=sys.stderr, flush=True)
 
     return result
 

compute/compute/snippets/images/create.py

@@ -63,14 +63,15 @@ def wait_for_extended_operation(
         print(
             f"Error during {verbose_name}: [Code: {operation.error_code}]: {operation.error_message}",
             file=sys.stderr,
+            flush=True,
         )
-        print(f"Operation ID: {operation.name}")
+        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)
+        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)
+            print(f" - {warning.code}: {warning.message}", file=sys.stderr, flush=True)
 
     return result