record deletion of images

Author: a-dubs

pycloudlib/azure/cloud.py

@@ -627,7 +627,7 @@ def delete_image(self, image_id, **kwargs):
         if delete_poller.status() == "Succeeded":
             if image_id in self.registered_images:
                 del self.registered_images[image_id]
-                self._log.debug("Image %s was deleted", image_id)
+                self._record_image_deletion(image_id)
         else:
             self._log.debug(
                 "Error deleting %s. Status: %d",

pycloudlib/cloud.py

@@ -41,10 +41,32 @@ class ImageType(enum.Enum):
 
 @dataclasses.dataclass
 class ImageInfo:
-    """Dataclass to hold image information."""
+    """Dataclass that represents an image on any given cloud."""
 
-    id: str
-    name: str
+    image_id: str
+    image_name: str
+
+    def __str__(self):
+        """Return a human readable string representation of the image."""
+        return f"{self.image_name} [id: {self.image_id}]"
+
+    def __repr__(self):
+        """Return a string representation of the image."""
+        return f"ImageInfo(id={self.image_id}, name={self.image_name})"
+
+    def __eq__(self, other):
+        """
+        Check if two ImageInfo objects represent the same image.
+
+        Only the id is used for comparison since this should be the unique identifier for an image.
+        """
+        if not isinstance(other, ImageInfo):
+            return False
+        return self.image_id == other.image_id
+
+    def __dict__(self):
+        """Return a dictionary representation of the image."""
+        return {"image_id": self.image_id, "image_name": self.image_name}
 
 
 class BaseCloud(ABC):
@@ -225,15 +247,15 @@ def clean(self) -> List[Exception]:
                 exceptions.append(e)
         for image_info in self.created_images:
             try:
-                self.delete_image(image_id=image_info.id)
+                self.delete_image(image_id=image_info.image_id)
             except Exception as e:
                 exceptions.append(e)
         for image_info in self.preserved_images:
             # noop - just log that we're not cleaning up these images
             self._log.info(
                 "Preserved image %s [id:%s] is NOT being cleaned up.",
-                image_info.name,
-                image_info.id,
+                image_info.image_name,
+                image_info.image_id,
             )
         return exceptions
 
@@ -329,6 +351,18 @@ def _validate_tag(tag: str):
             raise InvalidTagNameError(tag=tag, rules_failed=rules_failed)
 
     def _get_ssh_keys(self) -> KeyPair:
+        """
+        Get the ssh key pair to use for the cloud instance.
+
+        If no key pair is provided in the config file, the default key pair
+        will be used. The default key pair is the id_rsa or id_ed25519 key
+        in the user's .ssh directory.
+
+        :raises PycloudlibError: if no public key path is provided and no default key is found
+        :raises PycloudlibError: if the public key path provided in the config does not exist
+
+        :return: KeyPair object with the public and private key paths
+        """
         user = getpass.getuser()
         # check if id_rsa or id_ed25519 keys exist in the user's .ssh directory
         possible_default_keys = [
@@ -380,21 +414,48 @@ def _store_snapshot_info(
         :return: ImageInfo object with the snapshot information
         """
         image_info = ImageInfo(
-            id=snapshot_id,
-            name=snapshot_name,
+            image_id=snapshot_id,
+            image_name=snapshot_name,
         )
         if not keep_snapshot:
             self.created_images.append(image_info)
             self._log.info(
-                "Created temporary snapshot %s [id:%s]",
-                image_info.name,
-                image_info.id,
+                "Created temporary snapshot %s",
+                image_info,
             )
         else:
             self.preserved_images.append(image_info)
             self._log.info(
-                "Created permanent snapshot %s [id:%s]",
-                image_info.name,
-                image_info.id,
+                "Created permanent snapshot %s",
+                image_info,
             )
         return image_info
+
+    def _record_image_deletion(self, image_id: str):
+        """
+        Record the deletion of an image.
+
+        This method should be called after an image is successfully deleted.
+        It will remove the image from the list of created_images or preserved_images
+        so that the cloud does not attempt to re-clean it up later. It will also log
+        the deletion of the image.
+
+        :param image_id: ID of the image that was deleted
+        """
+        if match := [i for i in self.created_images if i.image_id == image_id]:
+            deleted_image = match[0]
+            self.created_images.remove(deleted_image)
+            self._log.debug(
+                "Snapshot %s has been deleted. Will no longer need to be cleaned up later.",
+                deleted_image,
+            )
+        elif match := [i for i in self.preserved_images if i.image_id == image_id]:
+            deleted_image = match[0]
+            self.preserved_images.remove(deleted_image)
+            self._log.debug(
+                "Snapshot %s has been deleted. This snapshot was taken with keep=True, "
+                "but since it has been manually deleted, it will not be preserved.",
+                deleted_image,
+            )
+        else:
+            self._log.debug("Deleted image %s", image_id)

pycloudlib/ec2/cloud.py

@@ -291,6 +291,8 @@ def delete_image(self, image_id, **kwargs):
         self._log.debug("removing custom snapshot %s", snapshot_id)
         self.client.delete_snapshot(SnapshotId=snapshot_id)
 
+        self._record_image_deletion(image_id)
+
     def delete_key(self, name):
         """Delete an uploaded key.
 

pycloudlib/gce/cloud.py

@@ -314,6 +314,7 @@ def delete_image(self, image_id, **kwargs):
             raise_on_error(operation)
         except GoogleAPICallError as e:
             raise_on_error(e)
+        self._record_image_deletion(image_id)
 
     def get_instance(
         self,

pycloudlib/ibm/cloud.py

@@ -13,7 +13,7 @@
 
 from pycloudlib.cloud import BaseCloud
 from pycloudlib.config import ConfigFile
-from pycloudlib.errors import InvalidTagNameError
+from pycloudlib.errors import InvalidTagNameError, ResourceNotFoundError, ResourceType
 from pycloudlib.ibm._util import get_first as _get_first
 from pycloudlib.ibm._util import iter_resources as _iter_resources
 from pycloudlib.ibm._util import wait_until as _wait_until
@@ -130,7 +130,9 @@ def delete_image(self, image_id: str, **kwargs):
             self._client.delete_image(image_id).get_result()
         except ApiException as e:
             if "does not exist" not in str(e):
-                raise
+                raise ResourceNotFoundError(ResourceType.IMAGE, image_id) from e
+        else:
+            self._record_image_deletion(image_id)
 
     def released_image(self, release, *, arch: str = "amd64", **kwargs):
         """ID of the latest released image for a particular release.

pycloudlib/ibm_classic/cloud.py

@@ -81,6 +81,8 @@ def delete_image(self, image_id: str, **kwargs):
             ) from e
         except SoftLayer.SoftLayerAPIError as e:
             raise IBMClassicException(f"Error deleting image {image_id}") from e
+        else:
+            self._record_image_deletion(image_id)
 
     def released_image(self, release, *, disk_size: str = "25G", **kwargs):
         """ID (globalIdentifier) of the latest released image for a particular release.

pycloudlib/lxd/cloud.py

@@ -393,11 +393,10 @@ def delete_image(self, image_id, **kwargs):
             image_id: string, LXD image fingerprint
         """
         self._log.debug("Deleting image: '%s'", image_id)
-
         subp(["lxc", "image", "delete", image_id])
-        self._log.debug("Deleted %s", image_id)
+        self._record_image_deletion(image_id)
 
-    def snapshot(self, instance, *, clean=True, keep=False, name=None):
+    def snapshot(self, instance: LXDInstance, *, clean=True, keep=False, name=None):  # type: ignore
         """Take a snapshot of the passed in instance for use as image.
 
         :param instance: The instance to create an image from
@@ -428,13 +427,6 @@ def clean(self) -> List[Exception]:
         """
         exceptions = super().clean()
 
-        for snapshot in self.created_snapshots:
-            try:
-                subp(["lxc", "image", "delete", snapshot])
-            except RuntimeError as e:
-                if "Image not found" not in str(e):
-                    exceptions.append(e)
-
         for profile in self.created_profiles:
             try:
                 subp(["lxc", "profile", "delete", profile])

pycloudlib/oci/cloud.py

@@ -126,6 +126,7 @@ def delete_image(self, image_id, **kwargs):
             image_id: string, id of the image to delete
         """
         self.compute_client.delete_image(image_id, **kwargs)
+        self._record_image_deletion(image_id)
 
     def released_image(self, release, operating_system="Canonical Ubuntu"):
         """Get the released image.

pycloudlib/openstack/cloud.py

@@ -56,6 +56,7 @@ def delete_image(self, image_id, **kwargs):
             image_id: string, id of the image to delete
         """
         self.conn.delete_image(image_id, wait=True)
+        self._record_image_deletion(image_id)
 
     def released_image(self, release, **kwargs):
         """Not supported for openstack."""

pycloudlib/qemu/cloud.py

@@ -107,6 +107,7 @@ def delete_image(self, image_id, **kwargs):
         image_file = Path(image_id)
         if image_file.exists():
             image_file.unlink()
+            self._record_image_deletion(image_id)
         else:
             self._log.debug("Cannot delete image %s as it does not exist", image_file)