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

samples/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>

samples/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)

samples/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>
+
+

samples/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>

samples/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>

samples/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>

samples/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
 

samples/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
 

samples/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
 

samples/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