delete-project subcommand

src/devbox/cli.py

@@ -59,12 +59,10 @@ def terminate(ctx, instance_id: str):
     manager = ctx.obj['manager']
 
     try:
-        success, message = manager.terminate_instance(instance_id, console)
-        if success:
-            console.print_success(message)
-        else:
-            console.print_error(message)
-            sys.exit(1)
+        result = manager.terminate_instance(instance_id, console)
+        console.print_success(
+            f"Terminating instance {result['instance_id']} (project: {result['project']})."
+        )
     except Exception as e:
         console.print_error(f"Failed to terminate instance: {str(e)}")
         sys.exit(1)
@@ -124,6 +122,53 @@ def new(ctx, project: str, base_ami: str, param_prefix: str):
         console.print_error(f"Failed to create project: {str(e)}")
         sys.exit(1)
 
+@cli.command()
+@click.argument('project')
+@click.option('--force', is_flag=True, help='Skip confirmation prompts')
+@click.pass_context
+def delete_project(ctx, project: str, force: bool):
+    """Delete a DevBox project and its AMI/snapshots."""
+    console = ctx.obj['console']
+    manager = ctx.obj['manager']
+
+    try:
+        item = manager.get_project_item(project)
+        if not item:
+            console.print_error(f"Project '{project}' not found in the main table.")
+            sys.exit(1)
+
+        in_use, reason = manager.project_in_use(project, item)
+        if in_use:
+            console.print_error(f"Project '{project}' is currently in use: {reason}")
+            sys.exit(1)
+
+        if not force:
+            if not click.confirm(f"Delete project '{project}' from the main table?"):
+                console.print_warning("Project deletion cancelled.")
+                return
+
+        manager.delete_project_entry(project)
+        console.print_success(f"Deleted project '{project}' from the main table.")
+
+        ami_id = item.get("AMI")
+        if not ami_id:
+            console.print_warning(f"No AMI recorded for project '{project}'.")
+            return
+
+        if not force:
+            if not click.confirm(f"Also delete AMI {ami_id} and its backing snapshot(s)?"):
+                console.print_warning("AMI cleanup cancelled.")
+                return
+
+        result = manager.delete_ami_and_snapshots(ami_id)
+        console.print_success(
+            f"Deregistered AMI {result['ami_id']} and deleted {result['snapshot_count']} snapshot(s)."
+        )
+
+    except Exception as e:
+        console.print_error(f"Failed to delete project: {str(e)}")
+        sys.exit(1)
+
 def main():
     """Entry point for the CLI."""
     cli(obj={})

src/devbox/devbox_manager.py

@@ -203,22 +203,149 @@ def list_snapshots(self, project: Optional[str] = None, console=None, orphan_onl
                 original_exception=e
             )
 
-    def terminate_instance(self, identifier: str, console=None) -> Tuple[bool, str]:
+    def get_project_item(self, project: str) -> Optional[Dict[str, Any]]:
+        """Get a project entry from the main DynamoDB table.
+
+        Args:
+            project: Project name
+
+        Returns:
+            Project item if it exists, otherwise None
+        """
+        try:
+            table = self.get_table()
+            response = table.get_item(Key={"project": project})
+            return response.get("Item")
+        except ClientError as e:
+            raise utils.AWSClientError(
+                f"Failed to retrieve project '{project}': {str(e)}",
+                error_code=e.response.get('Error', {}).get('Code'),
+                original_exception=e
+            )
+
+    def project_in_use(self, project: str, item: Optional[Dict[str, Any]] = None) -> Tuple[bool, str]:
+        """Check whether a project is currently in use.
+
+        Args:
+            project: Project name
+            item: Optional project item from DynamoDB
+
+        Returns:
+            Tuple of (in_use, reason)
+        """
+        try:
+            filters = [
+                {"Name": "tag:Project", "Values": [project]},
+                {"Name": "instance-state-name", "Values": ["pending", "running", "stopping", "stopped", "shutting-down"]}
+            ]
+            response = self.ec2.describe_instances(Filters=filters)
+            instances = [
+                instance
+                for reservation in response.get("Reservations", [])
+                for instance in reservation.get("Instances", [])
+            ]
+            if instances:
+                states = sorted({i.get("State", {}).get("Name", "unknown") for i in instances})
+                return True, f"EC2 instances in states: {', '.join(states)}."
+        except ClientError as e:
+            raise utils.AWSClientError(
+                f"Failed to check instance usage for project '{project}': {str(e)}",
+                error_code=e.response.get('Error', {}).get('Code'),
+                original_exception=e
+            )
+
+        status = (item or {}).get("Status")
+        in_use_statuses = {"LAUNCHING", "SNAPSHOTTING", "IMAGING"}
+        if status in in_use_statuses:
+            return True, f"Project status is {status}."
+
+        return False, ""
+
+    def delete_project_entry(self, project: str) -> None:
+        """Delete a project entry from the main DynamoDB table.
+
+        Args:
+            project: Project name
+        """
+        try:
+            table = self.get_table()
+            table.delete_item(Key={"project": project})
+        except ClientError as e:
+            raise utils.AWSClientError(
+                f"Failed to delete project '{project}' from the main table: {str(e)}",
+                error_code=e.response.get('Error', {}).get('Code'),
+                original_exception=e
+            )
+
+    def delete_ami_and_snapshots(self, ami_id: str) -> Dict[str, Any]:
+        """Deregister an AMI and delete its backing snapshots.
+
+        Args:
+            ami_id: AMI ID to delete
+
+        Returns:
+            Dict with AMI cleanup details
+        """
+        if not ami_id:
+            raise ValueError("No AMI ID provided.")
+
+        try:
+            response = self.ec2.describe_images(ImageIds=[ami_id])
+            images = response.get("Images", [])
+            if not images:
+                raise utils.ResourceNotFoundError(f"AMI {ami_id} not found.")
+
+            image = images[0]
+            snapshot_ids = []
+            for mapping in image.get("BlockDeviceMappings", []):
+                ebs = mapping.get("Ebs", {})
+                snapshot_id = ebs.get("SnapshotId")
+                if snapshot_id:
+                    snapshot_ids.append(snapshot_id)
+
+            self.ec2.deregister_image(ImageId=ami_id)
+
+            failed_snapshots = []
+            for snapshot_id in snapshot_ids:
+                try:
+                    self.ec2.delete_snapshot(SnapshotId=snapshot_id)
+                except ClientError as e:
+                    failed_snapshots.append((snapshot_id, e))
+
+            if failed_snapshots:
+                failed_ids = ", ".join(snap_id for snap_id, _ in failed_snapshots)
+                raise utils.DevBoxError(
+                    f"Deregistered AMI {ami_id} but failed to delete snapshots: {failed_ids}."
+                )
+
+            return {"ami_id": ami_id, "snapshot_count": len(snapshot_ids)}
+
+        except ClientError as e:
+            error_code = e.response.get('Error', {}).get('Code', 'UnknownError')
+            raise utils.AWSClientError(
+                f"Error deleting AMI {ami_id}: {error_code} - {str(e)}",
+                error_code=error_code,
+                original_exception=e
+            )
+
+    def terminate_instance(self, identifier: str, console=None) -> Dict[str, str]:
         """Terminate an instance by ID or project name.
 
         Args:
             identifier: Either an instance ID or project name
             console: Optional console output handler
 
         Returns:
-            A tuple of (success: bool, message: str)
+            Dict with instance termination details
         """
         try:
             # First, try to find instances by project name
             instances = self.list_instances(project=identifier)
 
             if len(instances) > 1:
-                return False, f"Multiple instances found for project '{identifier}'. Please specify instance ID instead."
+                raise utils.DevBoxError(
+                    f"Multiple instances found for project '{identifier}'. Please specify instance ID instead."
+                )
             elif len(instances) == 1:
                 instance_id = instances[0]['InstanceId']
                 project = instances[0]['Project']
@@ -230,14 +357,22 @@ def terminate_instance(self, identifier: str, console=None) -> Tuple[bool, str]:
                     instance_id = instance['InstanceId']
                     project = utils.get_project_tag(instance.get('Tags', []))
                     if not project:
-                        return False, f"Instance {identifier} is not managed by devbox (missing Project tag)."
+                        raise utils.DevBoxError(
+                            f"Instance {identifier} is not managed by devbox (missing Project tag)."
+                        )
                 except (ClientError, KeyError, IndexError):
-                    return False, f"No instance found with ID or project name: {identifier}"
+                    raise utils.ResourceNotFoundError(
+                        f"No instance found with ID or project name: {identifier}"
+                    )
 
             # Terminate the instance
             self.ec2.terminate_instances(InstanceIds=[instance_id])
-            return True, f"Terminating instance {instance_id} (project: {project})."
+            return {"instance_id": instance_id, "project": project}
 
         except ClientError as e:
             error_code = e.response.get('Error', {}).get('Code', 'UnknownError')
-            return False, f"Error terminating instance: {error_code} - {str(e)}"
+            raise utils.AWSClientError(
+                f"Error terminating instance: {error_code} - {str(e)}",
+                error_code=error_code,
+                original_exception=e
+            )