diff --git a/compute/compute/ingredients/images/create.py b/compute/compute/ingredients/images/create.py
new file mode 100644
index 000000000000..805f1e9c7b48
--- /dev/null
+++ b/compute/compute/ingredients/images/create.py
@@ -0,0 +1,83 @@
+#  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
+import time
+
+from google.cloud import compute_v1
+import warnings
+
+# <INGREDIENT create_image>
+STOPPED_MACHINE_STATUS = (
+    compute_v1.Instance.Status.TERMINATED.name,
+    compute_v1.Instance.Status.STOPPED.name
+)
+
+
+def create_image(project_id: str, zone: str, source_disk_name: str, image_name: str,
+                 storage_location: str = None, force_create: bool = False) -> compute_v1.Image:
+    """
+    Creates a new disk image.
+
+    Args:
+        project_id: project ID or project number of the Cloud project you use.
+        zone: zone of the disk you copy from.
+        source_disk_name: name of the source disk you copy from.
+        image_name: name of the image you want to create.
+        storage_location: storage location for the image. If the value is undefined,
+            function will store the image in the multi-region closest to your image's
+            source location.
+        force_create: create the image even if the source disk is attached to a
+            running instance.
+    """
+    image_client = compute_v1.ImagesClient()
+    disk_client = compute_v1.DisksClient()
+    instance_client = compute_v1.InstancesClient()
+
+    # Get source disk
+    disk = disk_client.get(project=project_id, zone=zone, disk=source_disk_name)
+
+    for disk_user in disk.users:
+        instance = instance_client.get(project=project_id, zone=zone, instance=disk_user)
+        if instance.status in STOPPED_MACHINE_STATUS:
+            continue
+        if not force_create:
+            raise RuntimeError(f"Instance {disk_user} should be stopped. For Windows instances please "
+                               f"stop the instance using `GCESysprep` command. For Linux instances just "
+                               f"shut it down normally. You can supress this error and create an image of"
+                               f"the disk by setting `force_create` parameter to true (not recommended). \n"
+                               f"More information here: \n"
+                               f" * https://cloud.google.com/compute/docs/instances/windows/creating-windows-os-image#api \n"
+                               f" * https://cloud.google.com/compute/docs/images/create-delete-deprecate-private-images#prepare_instance_for_image")
+        else:
+            warnings.warn(f"Warning: The `force_create` option may compromise the integrity of your image. "
+                          f"Stop the {disk_user} instance before you create the image if possible.")
+
+    # Create image
+    image = compute_v1.Image()
+    image.source_disk = disk.self_link
+    image.name = image_name
+    if storage_location:
+        image.storage_locations = [storage_location]
+
+    operation = image_client.insert(project=project_id, image_resource=image)
+
+    wait_for_extended_operation(operation, "image creation")
+
+    return image_client.get(project=project_id, image=image_name)
+# </INGREDIENT>
diff --git a/compute/compute/ingredients/images/delete.py b/compute/compute/ingredients/images/delete.py
new file mode 100644
index 000000000000..18059fd5348f
--- /dev/null
+++ b/compute/compute/ingredients/images/delete.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.
+
+
+# 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 NoReturn
+
+from google.cloud import compute_v1
+
+
+# <INGREDIENT delete_image>
+def delete_image(project_id: str, image_name: str) -> NoReturn:
+    """
+    Deletes a disk image.
+
+    Args:
+        project_id: project ID or project number of the Cloud project you use.
+        image_name: name of the image you want to delete.
+    """
+    image_client = compute_v1.ImagesClient()
+    operation = image_client.delete(project=project_id, image=image_name)
+    wait_for_extended_operation(operation, "image deletion")
+# </INGREDIENT>
diff --git a/compute/compute/recipes/images/create.py b/compute/compute/recipes/images/create.py
new file mode 100644
index 000000000000..22c0d19cc502
--- /dev/null
+++ b/compute/compute/recipes/images/create.py
@@ -0,0 +1,24 @@
+#  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_windows_image_create>
+# <REGION compute_images_create>
+# <IMPORTS/>
+
+# <INGREDIENT wait_for_extended_operation />
+
+# <INGREDIENT create_image />
+# </REGION compute_images_create>
+# </REGION compute_windows_image_create>
diff --git a/compute/compute/recipes/images/delete.py b/compute/compute/recipes/images/delete.py
new file mode 100644
index 000000000000..6796dae9eea8
--- /dev/null
+++ b/compute/compute/recipes/images/delete.py
@@ -0,0 +1,22 @@
+#  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_images_delete>
+# <IMPORTS/>
+
+# <INGREDIENT wait_for_extended_operation />
+
+# <INGREDIENT delete_image />
+# </REGION compute_images_delete>
diff --git a/compute/compute/snippets/images/create.py b/compute/compute/snippets/images/create.py
new file mode 100644
index 000000000000..af2a96bd8118
--- /dev/null
+++ b/compute/compute/snippets/images/create.py
@@ -0,0 +1,151 @@
+#  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
+
+
+# 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_windows_image_create]
+# [START compute_images_create]
+import sys
+import time
+from typing import Any
+import warnings
+
+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:
+    """
+    This method will wait 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,
+        )
+        print(f"Operation ID: {operation.name}")
+        raise operation.exception() or RuntimeError(operation.error_message)
+
+    if operation.warnings:
+        print(f"Warnings during {verbose_name}:\n", file=sys.stderr)
+        for warning in operation.warnings:
+            print(f" - {warning.code}: {warning.message}", file=sys.stderr)
+
+    return result
+
+
+STOPPED_MACHINE_STATUS = (
+    compute_v1.Instance.Status.TERMINATED.name,
+    compute_v1.Instance.Status.STOPPED.name,
+)
+
+
+def create_image(
+    project_id: str,
+    zone: str,
+    source_disk_name: str,
+    image_name: str,
+    storage_location: str = None,
+    force_create: bool = False,
+) -> compute_v1.Image:
+    """
+    Creates a new disk image.
+
+    Args:
+        project_id: project ID or project number of the Cloud project you use.
+        zone: zone of the disk you copy from.
+        source_disk_name: name of the source disk you copy from.
+        image_name: name of the image you want to create.
+        storage_location: storage location for the image. If the value is undefined,
+            function will store the image in the multi-region closest to your image's
+            source location.
+        force_create: create the image even if the source disk is attached to a
+            running instance.
+    """
+    image_client = compute_v1.ImagesClient()
+    disk_client = compute_v1.DisksClient()
+    instance_client = compute_v1.InstancesClient()
+
+    # Get source disk
+    disk = disk_client.get(project=project_id, zone=zone, disk=source_disk_name)
+
+    for disk_user in disk.users:
+        instance = instance_client.get(
+            project=project_id, zone=zone, instance=disk_user
+        )
+        if instance.status in STOPPED_MACHINE_STATUS:
+            continue
+        if not force_create:
+            raise RuntimeError(
+                f"Instance {disk_user} should be stopped. For Windows instances please "
+                f"stop the instance using `GCESysprep` command. For Linux instances just "
+                f"shut it down normally. You can supress this error and create an image of"
+                f"the disk by setting `force_create` parameter to true (not recommended). \n"
+                f"More information here: \n"
+                f" * https://cloud.google.com/compute/docs/instances/windows/creating-windows-os-image#api \n"
+                f" * https://cloud.google.com/compute/docs/images/create-delete-deprecate-private-images#prepare_instance_for_image"
+            )
+        else:
+            warnings.warn(
+                f"Warning: The `force_create` option may compromise the integrity of your image. "
+                f"Stop the {disk_user} instance before you create the image if possible."
+            )
+
+    # Create image
+    image = compute_v1.Image()
+    image.source_disk = disk.self_link
+    image.name = image_name
+    if storage_location:
+        image.storage_locations = [storage_location]
+
+    operation = image_client.insert(project=project_id, image_resource=image)
+
+    wait_for_extended_operation(operation, "image creation")
+
+    return image_client.get(project=project_id, image=image_name)
+
+
+# [END compute_images_create]
+# [END compute_windows_image_create]
diff --git a/compute/compute/snippets/images/delete.py b/compute/compute/snippets/images/delete.py
new file mode 100644
index 000000000000..7216674a2e1d
--- /dev/null
+++ b/compute/compute/snippets/images/delete.py
@@ -0,0 +1,89 @@
+#  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
+
+
+# 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_images_delete]
+import sys
+from typing import Any, NoReturn
+
+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:
+    """
+    This method will wait 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,
+        )
+        print(f"Operation ID: {operation.name}")
+        raise operation.exception() or RuntimeError(operation.error_message)
+
+    if operation.warnings:
+        print(f"Warnings during {verbose_name}:\n", file=sys.stderr)
+        for warning in operation.warnings:
+            print(f" - {warning.code}: {warning.message}", file=sys.stderr)
+
+    return result
+
+
+def delete_image(project_id: str, image_name: str) -> NoReturn:
+    """
+    Deletes a disk image.
+
+    Args:
+        project_id: project ID or project number of the Cloud project you use.
+        image_name: name of the image you want to delete.
+    """
+    image_client = compute_v1.ImagesClient()
+    operation = image_client.delete(project=project_id, image=image_name)
+    wait_for_extended_operation(operation, "image deletion")
+
+
+# [END compute_images_delete]
diff --git a/compute/compute/snippets/tests/test_images.py b/compute/compute/snippets/tests/test_images.py
index 18852ac09a08..394114b29723 100644
--- a/compute/compute/snippets/tests/test_images.py
+++ b/compute/compute/snippets/tests/test_images.py
@@ -11,10 +11,37 @@
 # 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.
+import uuid
 
+import google.auth
+from google.cloud import compute_v1
+import pytest
+
+from ..disks.create_from_image import create_disk_from_image
+from ..disks.delete import delete_disk
+from ..images.create import create_image
+from ..images.delete import delete_image
 from ..images.get import get_image
+from ..images.get import get_image_from_family
 from ..images.list import list_images
 
+PROJECT = google.auth.default()[1]
+ZONE = 'europe-central2-c'
+
+
+@pytest.fixture
+def test_disk():
+    """
+    Get the newest version of debian 11 and make a disk from it.
+    """
+    new_debian = get_image_from_family('debian-cloud', 'debian-11')
+    test_disk_name = "test-disk-" + uuid.uuid4().hex[:10]
+    disk = create_disk_from_image(PROJECT, ZONE, test_disk_name,
+                                  f"zones/{ZONE}/diskTypes/pd-standard",
+                                  20, new_debian.self_link)
+    yield disk
+    delete_disk(PROJECT, ZONE, test_disk_name)
+
 
 def test_list_images():
     images = list_images("debian-cloud")
@@ -32,3 +59,18 @@ def test_get_image():
     image2 = get_image("debian-cloud", image.name)
 
     assert image == image2
+
+
+def test_create_delete_image(test_disk):
+    test_image_name = "test-image-" + uuid.uuid4().hex[:10]
+    new_image = create_image(PROJECT, ZONE, test_disk.name, test_image_name)
+    try:
+        assert new_image.name == test_image_name
+        assert new_image.disk_size_gb == 20
+        assert isinstance(new_image, compute_v1.Image)
+    finally:
+        delete_image(PROJECT, test_image_name)
+
+        for image in list_images(PROJECT):
+            if image.name == test_image_name:
+                pytest.fail(f"Image {test_image_name} should have been deleted.")