Add confirmation prompt for destructive bulk Admin API methods

.gitignore

@@ -10,3 +10,5 @@ venv
 .idea
 
 .cld-sync
+.cld-settings
+.venv

cloudinary_cli/core/admin.py

@@ -21,7 +21,8 @@
         help="Pass optional parameters as interpreted strings.")
 @option("-A", "--auto_paginate", is_flag=True, help="Will auto paginate Admin API calls.", default=False)
 @option("-ff", "--filter_fields", multiple=True, help="Filter fields to return when using auto pagination.")
-@option("-F", "--force", is_flag=True, help="Skip confirmation when running --auto-paginate.")
+@option("-F", "--force", is_flag=True,
+        help="Skip confirmations for auto pagination and destructive bulk API methods.")
 @option("-ls", "--ls", is_flag=True, help="List all available methods in the Admin API.")
 @option("--save", nargs=1, help="Save output to a file.")
 @option("-d", "--doc", is_flag=True, help="Open the Admin API reference in a browser.")

cloudinary_cli/utils/api_utils.py

@@ -20,6 +20,37 @@
 
 _cursor_fields = {"resource": "derived_next_cursor"}
 
+# Selector-style destructive bulk Admin API methods.
+# These can match an open-ended set of assets without the caller enumerating each target,
+# so an interactive confirmation is required (bypassable via --force).
+DESTRUCTIVE_BULK_API_METHODS = {
+    "delete_all_resources",
+    "delete_resources_by_prefix",
+    "delete_resources_by_tag",
+}
+
+# Server-enforced cap on how many resources a single destructive bulk call can affect
+# (NConfig.max_resource_count_for_delete). Used purely for clearer prompt wording.
+MAX_DESTRUCTIVE_BULK_PER_CALL = 1000
+
+
+def is_destructive_bulk_api_method(method_name):
+    return method_name in DESTRUCTIVE_BULK_API_METHODS
+
+
+def confirm_destructive_bulk_api_method(api_name, method_name, force=False):
+    if force or not is_destructive_bulk_api_method(method_name):
+        return True
+
+    display_api_name = api_name.capitalize()
+    if not confirm_action(
+            f"This will delete up to {MAX_DESTRUCTIVE_BULK_PER_CALL} matching assets via "
+            f"{display_api_name} API method '{method_name}'. Continue? (y/N)"):
+        logger.info("Stopping.")
+        return False
+
+    return True
+
 
 def query_cld_folder(folder, folder_mode, status=None):
     files = {}
@@ -311,6 +342,9 @@ def handle_api_command(
         log_exception(e)
         return False
 
+    if not confirm_destructive_bulk_api_method(api_name, func.__name__, force):
+        return False
+
     if not is_valid_cloudinary_config():
         raise ConfigurationError("No Cloudinary configuration found.")
 

test/test_cli_api.py

@@ -11,6 +11,8 @@
 API_MOCK_RESPONSE = api_response_mock()
 UPLOAD_MOCK_RESPONSE = uploader_response_mock()
 
+CONFIRM_ACTION_PATCH = "cloudinary_cli.utils.api_utils.confirm_action"
+
 
 class TestCLIApi(unittest.TestCase):
     runner = CliRunner()
@@ -39,3 +41,87 @@ def test_provisioning(self, mocker):
 
         self.assertEqual(0, result.exit_code, result.output)
         self.assertIn('"foo": "bar"', result.output)
+
+
+class TestDestructiveBulkConfirmation(unittest.TestCase):
+    runner = CliRunner()
+
+    @patch(CONFIRM_ACTION_PATCH, return_value=False)
+    @patch(URLLIB3_REQUEST)
+    def test_delete_all_resources_decline_skips_call(self, http_mock, confirm_mock):
+        http_mock.return_value = API_MOCK_RESPONSE
+        result = self.runner.invoke(cli, ['admin', 'delete_all_resources'])
+
+        self.assertEqual(0, result.exit_code, result.output)
+        confirm_mock.assert_called_once()
+        self.assertFalse(http_mock.called, "SDK should not be called when user declines")
+
+    @patch(CONFIRM_ACTION_PATCH, return_value=True)
+    @patch(URLLIB3_REQUEST)
+    def test_delete_all_resources_accept_calls_sdk(self, http_mock, confirm_mock):
+        http_mock.return_value = API_MOCK_RESPONSE
+        result = self.runner.invoke(cli, ['admin', 'delete_all_resources'])
+
+        self.assertEqual(0, result.exit_code, result.output)
+        confirm_mock.assert_called_once()
+        self.assertTrue(http_mock.called, "SDK should be called when user accepts")
+
+    @patch(CONFIRM_ACTION_PATCH, return_value=False)
+    @patch(URLLIB3_REQUEST)
+    def test_delete_all_resources_force_skips_prompt(self, http_mock, confirm_mock):
+        http_mock.return_value = API_MOCK_RESPONSE
+        result = self.runner.invoke(cli, ['admin', '-F', 'delete_all_resources'])
+
+        self.assertEqual(0, result.exit_code, result.output)
+        self.assertFalse(confirm_mock.called, "--force should bypass the confirmation prompt")
+        self.assertTrue(http_mock.called, "SDK should be called when --force is set")
+
+    @patch(CONFIRM_ACTION_PATCH, return_value=False)
+    @patch(URLLIB3_REQUEST)
+    def test_delete_resources_by_tag_decline_skips_call(self, http_mock, confirm_mock):
+        http_mock.return_value = API_MOCK_RESPONSE
+        result = self.runner.invoke(cli, ['admin', 'delete_resources_by_tag', 'mytag'])
+
+        self.assertEqual(0, result.exit_code, result.output)
+        confirm_mock.assert_called_once()
+        self.assertFalse(http_mock.called, "SDK should not be called when user declines")
+
+    @patch(CONFIRM_ACTION_PATCH, return_value=False)
+    @patch(URLLIB3_REQUEST)
+    def test_delete_resources_explicit_ids_no_prompt(self, http_mock, confirm_mock):
+        http_mock.return_value = API_MOCK_RESPONSE
+        result = self.runner.invoke(cli, ['admin', 'delete_resources', 'public_id1', 'public_id2'])
+
+        self.assertEqual(0, result.exit_code, result.output)
+        self.assertFalse(confirm_mock.called, "Explicit-ID delete must not prompt")
+        self.assertTrue(http_mock.called, "SDK should be called for explicit-ID delete")
+
+    @patch(CONFIRM_ACTION_PATCH, return_value=False)
+    @patch(URLLIB3_REQUEST)
+    def test_uploader_add_tag_no_prompt(self, http_mock, confirm_mock):
+        http_mock.return_value = UPLOAD_MOCK_RESPONSE
+        result = self.runner.invoke(cli, ['uploader', 'add_tag', 'mytag', 'public_id1'])
+
+        self.assertEqual(0, result.exit_code, result.output)
+        self.assertFalse(confirm_mock.called, "Non-destructive bulk methods must not prompt")
+        self.assertTrue(http_mock.called, "SDK should be called for non-destructive bulk methods")
+
+    @patch(CONFIRM_ACTION_PATCH, return_value=False)
+    @patch(URLLIB3_REQUEST)
+    def test_admin_resources_read_no_prompt(self, http_mock, confirm_mock):
+        http_mock.return_value = API_MOCK_RESPONSE
+        result = self.runner.invoke(cli, ['admin', 'resources'])
+
+        self.assertEqual(0, result.exit_code, result.output)
+        self.assertFalse(confirm_mock.called, "Read commands must not prompt")
+        self.assertTrue(http_mock.called, "SDK should be called for read commands")
+
+    @patch(CONFIRM_ACTION_PATCH, return_value=False)
+    @patch(URLLIB3_REQUEST)
+    def test_admin_resources_read_with_force_no_prompt(self, http_mock, confirm_mock):
+        http_mock.return_value = API_MOCK_RESPONSE
+        result = self.runner.invoke(cli, ['admin', '-F', 'resources'])
+
+        self.assertEqual(0, result.exit_code, result.output)
+        self.assertFalse(confirm_mock.called, "Read commands must not prompt regardless of --force")
+        self.assertTrue(http_mock.called)