(api) add option to board delete route and logic to services

invokeai/app/api/routers/boards.py

@@ -71,11 +71,19 @@ async def update_board(
 @boards_router.delete("/{board_id}", operation_id="delete_board")
 async def delete_board(
     board_id: str = Path(description="The id of board to delete"),
+    include_images: bool = Path(
+        description="Permanently delete all images on the board", default=False
+    ),
 ) -> None:
     """Deletes a board"""
 
     try:
-        ApiDependencies.invoker.services.boards.delete(board_id=board_id)
+        if include_images:
+            ApiDependencies.invoker.services.images.delete_images_on_board(
+                board_id=board_id
+            )
+        else:
+            ApiDependencies.invoker.services.boards.delete(board_id=board_id)
     except Exception as e:
         # TODO: Does this need any exception handling at all?
         pass

invokeai/app/services/image_file_storage.py

@@ -85,8 +85,10 @@ class DiskImageFileStorage(ImageFileStorageBase):
         self.__cache_ids = Queue()
         self.__max_cache_size = 10  # TODO: get this from config
 
-        self.__output_folder: Path = output_folder if isinstance(output_folder, Path) else Path(output_folder)
-        self.__thumbnails_folder = self.__output_folder / 'thumbnails'
+        self.__output_folder: Path = (
+            output_folder if isinstance(output_folder, Path) else Path(output_folder)
+        )
+        self.__thumbnails_folder = self.__output_folder / "thumbnails"
 
         # Validate required output folders at launch
         self.__validate_storage_folders()
@@ -94,7 +96,7 @@ class DiskImageFileStorage(ImageFileStorageBase):
     def get(self, image_name: str) -> PILImageType:
         try:
             image_path = self.get_path(image_name)
-            
+
             cache_item = self.__get_cache(image_path)
             if cache_item:
                 return cache_item
@@ -155,7 +157,7 @@ class DiskImageFileStorage(ImageFileStorageBase):
     # TODO: make this a bit more flexible for e.g. cloud storage
     def get_path(self, image_name: str, thumbnail: bool = False) -> Path:
         path = self.__output_folder / image_name
-        
+
         if thumbnail:
             thumbnail_name = get_thumbnail_name(image_name)
             path = self.__thumbnails_folder / thumbnail_name
@@ -166,7 +168,7 @@ class DiskImageFileStorage(ImageFileStorageBase):
         """Validates the path given for an image or thumbnail."""
         path = path if isinstance(path, Path) else Path(path)
         return path.exists()
-    
+
     def __validate_storage_folders(self) -> None:
         """Checks if the required output folders exist and create them if they don't"""
         folders: list[Path] = [self.__output_folder, self.__thumbnails_folder]
@@ -179,7 +181,9 @@ class DiskImageFileStorage(ImageFileStorageBase):
     def __set_cache(self, image_name: Path, image: PILImageType):
         if not image_name in self.__cache:
             self.__cache[image_name] = image
-            self.__cache_ids.put(image_name)  # TODO: this should refresh position for LRU cache
+            self.__cache_ids.put(
+                image_name
+            )  # TODO: this should refresh position for LRU cache
             if len(self.__cache) > self.__max_cache_size:
                 cache_id = self.__cache_ids.get()
                 if cache_id in self.__cache:

invokeai/app/services/image_record_storage.py

@@ -94,6 +94,11 @@ class ImageRecordStorageBase(ABC):
         """Deletes an image record."""
         pass
 
+    @abstractmethod
+    def delete_many(self, image_names: list[str]) -> None:
+        """Deletes many image records."""
+        pass
+
     @abstractmethod
     def save(
         self,
@@ -385,6 +390,25 @@ class SqliteImageRecordStorage(ImageRecordStorageBase):
         finally:
             self._lock.release()
 
+    def delete_many(self, image_names: list[str]) -> None:
+        try:
+            placeholders = ",".join("?" for _ in image_names)
+
+            self._lock.acquire()
+
+            # Construct the SQLite query with the placeholders
+            query = f"DELETE FROM images WHERE id_column IN ({placeholders})"
+
+            # Execute the query with the list of IDs as parameters
+            self._cursor.execute(query, placeholders)
+
+            self._conn.commit()
+        except sqlite3.Error as e:
+            self._conn.rollback()
+            raise ImageRecordDeleteException from e
+        finally:
+            self._lock.release()
+
     def save(
         self,
         image_name: str,

invokeai/app/services/images.py

@@ -112,6 +112,11 @@ class ImageServiceABC(ABC):
         """Deletes an image."""
         pass
 
+    @abstractmethod
+    def delete_images_on_board(self, board_id: str):
+        """Deletes all images on a board."""
+        pass
+
 
 class ImageServiceDependencies:
     """Service dependencies for the ImageService."""
@@ -341,6 +346,28 @@ class ImageService(ImageServiceABC):
             self._services.logger.error("Problem deleting image record and file")
             raise e
 
+    def delete_images_on_board(self, board_id: str):
+        try:
+            images = self._services.board_image_records.get_images_for_board(board_id)
+            image_name_list = list(
+                map(
+                    lambda r: r.image_name,
+                    images.items,
+                )
+            )
+            for image_name in image_name_list:
+                self._services.image_files.delete(image_name)
+            self._services.image_records.delete_many(image_name_list)
+        except ImageRecordDeleteException:
+            self._services.logger.error(f"Failed to delete image records")
+            raise
+        except ImageFileDeleteException:
+            self._services.logger.error(f"Failed to delete image files")
+            raise
+        except Exception as e:
+            self._services.logger.error("Problem deleting image records and files")
+            raise e
+
     def _get_metadata(
         self, session_id: Optional[str] = None, node_id: Optional[str] = None
     ) -> Union[ImageMetadata, None]: