Merge pull request #1951 from dbcli/RW/batch-default-non-interactive

Make batch execution noninteractive by default

Author: rolandwalker

changelog.md

@@ -10,6 +10,12 @@ Breaking Changes
 * Remove support for deprecated SSH jump functionality.
 * Remove support for `my.cnf` vendor MySQL option files.
 * Remove support for `.myclirc` files in the current working directory.
+* Make `--batch` and `--execute` non-interactive by default.
+
+
+Features
+---------
+* Add `--warn-batch` flag, which is off by default.
 
 
 Documentation

mycli/main.py

@@ -201,6 +201,10 @@ class CliArgs:
         clickdc=None,
         help='Warn before running a destructive query.',
     )
+    warn_batch: bool = clickdc.option(
+        is_flag=True,
+        help='Warn before running a destructive query when executing a script.',
+    )
     local_infile: bool | None = clickdc.option(
         type=bool,
         is_flag=False,
@@ -236,9 +240,11 @@ class CliArgs:
         type=str,
         help='SQL script to execute in batch mode.',
     )
+    # deprecated 2026-06-20
     noninteractive: bool = clickdc.option(
         is_flag=True,
-        help="Don't prompt during batch input.  Recommended.",
+        hidden=True,
+        deprecated='See --warn-batch.',
     )
     format: str | None = clickdc.option(
         type=click.Choice(['default', 'csv', 'tsv', 'table']),

mycli/main_modes/batch.py

@@ -114,17 +114,16 @@ def dispatch_batch_statements(
         else:
             mycli.main_formatter.format_name = 'tsv'
 
-    warn_confirmed: bool | None = True
-    if not cli_args.noninteractive and mycli.destructive_warning and is_destructive(mycli.destructive_keywords, statements):
+    execution_confirmed: bool | None = True
+    if cli_args.warn_batch and is_destructive(mycli.destructive_keywords, statements):
         try:
             # this seems to work, even though we are reading from stdin above
             sys.stdin = open('/dev/tty')
-            # bug: the prompt will not be visible if stdout is redirected
-            warn_confirmed = confirm_destructive_query(mycli.destructive_keywords, statements)
+            execution_confirmed = confirm_destructive_query(mycli.destructive_keywords, statements)
         except (IOError, OSError) as e:
             mycli.logger.warning('Unable to open TTY as stdin.')
             raise e
-    if warn_confirmed:
+    if execution_confirmed:
         if cli_args.throttle > 0 and batch_counter >= 1:
             time.sleep(cli_args.throttle)
         mycli.run_query(statements, checkpoint=cli_args.checkpoint, new_line=True)

mycli/main_modes/execute.py

@@ -5,6 +5,9 @@
 
 import click
 
+from mycli.packages.interactive_utils import confirm_destructive_query
+from mycli.packages.sql_utils import is_destructive
+
 if TYPE_CHECKING:
     from mycli.client import MyCli
     from mycli.main import CliArgs
@@ -34,8 +37,19 @@ def main_execute_from_cli(mycli: 'MyCli', cli_args: 'CliArgs') -> int:
         else:
             mycli.main_formatter.format_name = 'tsv'
 
-        mycli.run_query(execute_sql, checkpoint=cli_args.checkpoint)
-        return 0
+        execution_confirmed: bool | None = True
+        if cli_args.warn_batch and is_destructive(mycli.destructive_keywords, execute_sql):
+            try:
+                sys.stdin = open('/dev/tty')
+                execution_confirmed = confirm_destructive_query(mycli.destructive_keywords, execute_sql)
+            except (IOError, OSError) as e:
+                mycli.logger.warning('Unable to open TTY as stdin.')
+                raise e
+        if execution_confirmed:
+            mycli.run_query(execute_sql, checkpoint=cli_args.checkpoint)
+            return 0
+        else:
+            return 1
     except Exception as e:
         click.secho(str(e), err=True, fg="red")
         return 1

mycli/packages/interactive_utils.py

@@ -36,7 +36,7 @@ def confirm_destructive_query(keywords: list[str], queries: str) -> bool | None:
     """
     prompt_text = "You're about to run a destructive command.\nDo you want to proceed? (y/n)"
     if is_destructive(keywords, queries) and sys.stdin.isatty():
-        return prompt(prompt_text, type=BOOLEAN_TYPE)
+        return prompt(prompt_text, type=BOOLEAN_TYPE, err=True)
     else:
         return None
 

test/pytests/test_interactive_utils.py

@@ -92,7 +92,7 @@ def fake_is_destructive(keywords: list[str], query: str) -> bool:
     assert prompt_calls == [
         (
             ("You're about to run a destructive command.\nDo you want to proceed? (y/n)",),
-            {'type': interactive_utils.BOOLEAN_TYPE},
+            {'type': interactive_utils.BOOLEAN_TYPE, 'err': True},
         )
     ]
 
@@ -120,7 +120,7 @@ def fake_is_destructive(keywords: list[str], query: str) -> bool:
     assert prompt_calls == [
         (
             ("You're about to run a destructive command.\nDo you want to proceed? (y/n)",),
-            {'type': interactive_utils.BOOLEAN_TYPE},
+            {'type': interactive_utils.BOOLEAN_TYPE, 'err': True},
         )
     ]
 

test/pytests/test_main.py

@@ -772,6 +772,8 @@ def test_help_strings_end_with_periods():
     """Make sure click options have help text that end with a period."""
     for param in click_entrypoint.params:
         if isinstance(param, click.core.Option):
+            if param.hidden:
+                continue
             assert hasattr(param, "help")
             assert param.help.endswith(".")
 

test/pytests/test_main_modes_batch.py

@@ -24,7 +24,7 @@
 @dataclass
 class DummyCliArgs:
     format: str = 'tsv'
-    noninteractive: bool = True
+    warn_batch: bool = False
     throttle: float = 0.0
     checkpoint: str | TextIOWrapper | None = None
     batch: str | None = None
@@ -428,7 +428,7 @@ def test_dispatch_batch_statements_sets_expected_output_format(
 
 def test_dispatch_batch_statements_confirms_destructive_queries_before_running(monkeypatch) -> None:
     mycli = DummyMyCli(destructive_warning=True)
-    cli_args = DummyCliArgs(noninteractive=False)
+    cli_args = DummyCliArgs(warn_batch=True)
     opened_tty = object()
 
     monkeypatch.setattr(batch_mode, 'is_destructive', lambda _keywords, _statement: True)
@@ -444,7 +444,7 @@ def test_dispatch_batch_statements_confirms_destructive_queries_before_running(m
 
 def test_dispatch_batch_statements_skips_query_when_destructive_confirmation_is_rejected(monkeypatch) -> None:
     mycli = DummyMyCli(destructive_warning=True)
-    cli_args = DummyCliArgs(noninteractive=False)
+    cli_args = DummyCliArgs(warn_batch=True)
 
     monkeypatch.setattr(batch_mode, 'is_destructive', lambda _keywords, _statement: True)
     monkeypatch.setattr(batch_mode, 'confirm_destructive_query', lambda _keywords, _statement: False)
@@ -458,7 +458,7 @@ def test_dispatch_batch_statements_skips_query_when_destructive_confirmation_is_
 
 def test_dispatch_batch_statements_raises_when_tty_cannot_be_opened(monkeypatch) -> None:
     mycli = DummyMyCli(destructive_warning=True)
-    cli_args = DummyCliArgs(noninteractive=False)
+    cli_args = DummyCliArgs(warn_batch=True)
 
     monkeypatch.setattr(batch_mode, 'is_destructive', lambda _keywords, _statement: True)
     monkeypatch.setattr(batch_mode, 'open', lambda _path: (_ for _ in ()).throw(OSError('tty unavailable')), raising=False)

test/pytests/test_main_modes_execute.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import builtins
 from dataclasses import dataclass
 from types import SimpleNamespace
 from typing import Any, cast
@@ -14,6 +15,7 @@ class DummyCliArgs:
     execute: str | None
     format: str = 'tsv'
     batch: str | None = None
+    warn_batch: bool = False
     checkpoint: str | None = None
 
 
@@ -22,11 +24,21 @@ class DummyFormatter:
     format_name: str | None = None
 
 
+class DummyLogger:
+    def __init__(self) -> None:
+        self.warning_calls: list[str] = []
+
+    def warning(self, message: str) -> None:
+        self.warning_calls.append(message)
+
+
 class DummyMyCli:
     def __init__(self, run_query_error: Exception | None = None) -> None:
         self.main_formatter = DummyFormatter()
         self.run_query_error = run_query_error
         self.ran_queries: list[tuple[str, str | None]] = []
+        self.destructive_keywords = ['drop']
+        self.logger = DummyLogger()
 
     def run_query(self, query: str, checkpoint: str | None = None) -> None:
         if self.run_query_error is not None:
@@ -125,3 +137,61 @@ def test_main_execute_from_cli_reports_query_errors(monkeypatch) -> None:
     assert mycli.main_formatter.format_name == 'ascii'
     assert mycli.ran_queries == []
     assert secho_calls == [('boom', True, 'red')]
+
+
+def test_main_execute_from_cli_confirms_destructive_query(monkeypatch) -> None:
+    mycli = DummyMyCli()
+    tty = object()
+    confirm_calls: list[tuple[list[str], str]] = []
+
+    monkeypatch.setattr(execute_mode, 'sys', fake_sys(stdin_tty=True))
+    monkeypatch.setattr(execute_mode, 'is_destructive', lambda keywords, query: True)
+    monkeypatch.setattr(builtins, 'open', lambda path: tty)
+
+    def confirm_destructive_query(keywords: list[str], query: str) -> bool:
+        confirm_calls.append((keywords, query))
+        return True
+
+    monkeypatch.setattr(execute_mode, 'confirm_destructive_query', confirm_destructive_query)
+
+    result = main_execute_from_cli(mycli, DummyCliArgs(execute='drop table t', warn_batch=True))
+
+    assert result == 0
+    assert execute_mode.sys.stdin is tty
+    assert confirm_calls == [(['drop'], 'drop table t')]
+    assert mycli.ran_queries == [('drop table t', None)]
+
+
+def test_main_execute_from_cli_returns_error_when_destructive_query_is_rejected(monkeypatch) -> None:
+    mycli = DummyMyCli()
+
+    monkeypatch.setattr(execute_mode, 'sys', fake_sys(stdin_tty=True))
+    monkeypatch.setattr(execute_mode, 'is_destructive', lambda keywords, query: True)
+    monkeypatch.setattr(builtins, 'open', lambda path: object())
+    monkeypatch.setattr(execute_mode, 'confirm_destructive_query', lambda keywords, query: False)
+
+    result = main_execute_from_cli(mycli, DummyCliArgs(execute='drop table t', warn_batch=True))
+
+    assert result == 1
+    assert mycli.ran_queries == []
+
+
+def test_main_execute_from_cli_reports_tty_open_error_for_destructive_query(monkeypatch) -> None:
+    secho_calls: list[tuple[str, bool, str]] = []
+    mycli = DummyMyCli()
+
+    monkeypatch.setattr(execute_mode, 'sys', fake_sys(stdin_tty=True))
+    monkeypatch.setattr(execute_mode, 'is_destructive', lambda keywords, query: True)
+    monkeypatch.setattr(builtins, 'open', lambda path: (_ for _ in ()).throw(OSError('no tty')))
+    monkeypatch.setattr(
+        execute_mode.click,
+        'secho',
+        lambda message, err, fg: secho_calls.append((message, err, fg)),
+    )
+
+    result = main_execute_from_cli(mycli, DummyCliArgs(execute='drop table t', warn_batch=True))
+
+    assert result == 1
+    assert mycli.logger.warning_calls == ['Unable to open TTY as stdin.']
+    assert mycli.ran_queries == []
+    assert secho_calls == [('no tty', True, 'red')]