Add --batch option as an alternative to STDIN

changelog.md

@@ -1,7 +1,12 @@
-1.65.2 (2026/03/19)
+Upcoming (TBD)
 ==============
 
-Security
+Features
+---------
+* Add a `--batch` option as an alternative to STDIN.
+
+
+Internal
 --------
 * Harden `codex-review` workflow against script injection from untrusted PR metadata.
 

mycli/main.py

@@ -89,6 +89,7 @@
 from mycli.key_bindings import mycli_bindings
 from mycli.lexer import MyCliLexer
 from mycli.packages import special
+from mycli.packages.batch_utils import statements_from_filehandle
 from mycli.packages.checkup import do_checkup
 from mycli.packages.filepaths import dir_path_exists, guess_socket_location
 from mycli.packages.hybrid_redirection import get_redirect_components, is_redirect_command
@@ -119,7 +120,6 @@
 DEFAULT_WIDTH = 80
 DEFAULT_HEIGHT = 25
 MIN_COMPLETION_TRIGGER = 1
-MAX_MULTILINE_BATCH_STATEMENT = 5000
 EMPTY_PASSWORD_FLAG_SENTINEL = -1
 
 
@@ -2002,6 +2002,7 @@ def get_last_query(self) -> str | None:
     "--password-file", type=click.Path(), help="File or FIFO path containing the password to connect to the db if not specified otherwise."
 )
 @click.argument("database", default=None, nargs=1)
+@click.option('--batch', 'batch_file', type=str, help='SQL script to execute in batch mode.')
 @click.option("--noninteractive", is_flag=True, help="Don't prompt during batch input.  Recommended.")
 @click.option(
     '--format', 'batch_format', type=click.Choice(['default', 'csv', 'tsv', 'table']), help='Format for batch or --execute output.'
@@ -2071,6 +2072,7 @@ def cli(
     character_set: str | None,
     password_file: str | None,
     noninteractive: bool,
+    batch_file: str | None,
     batch_format: str | None,
     throttle: float,
     use_keyring_cli_opt: str | None,
@@ -2494,6 +2496,10 @@ def get_password_from_file(password_file: str | None) -> str | None:
 
     #  --execute argument
     if execute:
+        if not sys.stdin.isatty():
+            click.secho('Ignoring STDIN since --execute was also given.', err=True, fg='red')
+        if batch_file:
+            click.secho('Ignoring --batch since --execute was also given.', err=True, fg='red')
         try:
             if batch_format == 'csv':
                 mycli.main_formatter.format_name = 'csv'
@@ -2556,38 +2562,26 @@ def dispatch_batch_statements(statements: str, batch_counter: int) -> None:
             click.secho(str(e), err=True, fg="red")
             sys.exit(1)
 
-    if sys.stdin.isatty():
-        mycli.run_cli()
-    else:
-        stdin = click.get_text_stream("stdin")
-        statements = ''
-        line_counter = 0
-        batch_counter = 0
-        for stdin_text in stdin:
-            line_counter += 1
-            if line_counter > MAX_MULTILINE_BATCH_STATEMENT:
-                click.secho(
-                    f'Saw single input statement greater than {MAX_MULTILINE_BATCH_STATEMENT} lines; assuming a parsing error.',
-                    err=True,
-                    fg="red",
-                )
-                sys.exit(1)
-            statements += stdin_text
+    if batch_file or not sys.stdin.isatty():
+        if batch_file:
+            if not sys.stdin.isatty() and batch_file != '-':
+                click.secho('Ignoring STDIN since --batch was also given.', err=True, fg='red')
             try:
-                tokens = sqlglot.tokenize(statements, read='mysql')
-                if not tokens:
-                    continue
-                # we don't handle changing the delimiter within the batch input
-                if tokens[-1].text == ';':
-                    dispatch_batch_statements(statements, batch_counter)
-                    batch_counter += 1
-                    statements = ''
-                    line_counter = 0
-            except sqlglot.errors.TokenError:
-                continue
-        if statements:
-            dispatch_batch_statements(statements, batch_counter)
+                batch_h = click.open_file(batch_file)
+            except (OSError, FileNotFoundError):
+                click.secho(f'Failed to open --batch file: {batch_file}', err=True, fg='red')
+                sys.exit(1)
+        else:
+            batch_h = click.get_text_stream('stdin')
+        try:
+            for statement, counter in statements_from_filehandle(batch_h):
+                dispatch_batch_statements(statement, counter)
+        except ValueError as e:
+            click.secho(str(e), err=True, fg='red')
+            sys.exit(1)
         sys.exit(0)
+
+    mycli.run_cli()
     mycli.close()
 
 

mycli/packages/batch_utils.py

@@ -0,0 +1,30 @@
+from typing import IO, Generator
+
+import sqlglot
+
+MAX_MULTILINE_BATCH_STATEMENT = 5000
+
+
+def statements_from_filehandle(file_h: IO) -> Generator[tuple[str, int], None, None]:
+    statements = ''
+    line_counter = 0
+    batch_counter = 0
+    for batch_text in file_h:
+        line_counter += 1
+        if line_counter > MAX_MULTILINE_BATCH_STATEMENT:
+            raise ValueError(f'Saw single input statement greater than {MAX_MULTILINE_BATCH_STATEMENT} lines; assuming a parsing error.')
+        statements += batch_text
+        try:
+            tokens = sqlglot.tokenize(statements, read='mysql')
+            if not tokens:
+                continue
+            # we don't yet handle changing the delimiter within the batch input
+            if tokens[-1].text == ';':
+                yield (statements, batch_counter)
+                batch_counter += 1
+                statements = ''
+                line_counter = 0
+        except sqlglot.errors.TokenError:
+            continue
+    if statements:
+        yield (statements, batch_counter)

test/test_main.py

@@ -1256,6 +1256,123 @@ def test_execute_with_logfile(executor):
         print(f"An error occurred while attempting to delete the file: {e}")
 
 
+def _noninteractive_mock_mycli(monkeypatch):
+    class Formatter:
+        format_name = None
+
+    class Logger:
+        def debug(self, *args, **args_dict):
+            pass
+
+        def error(self, *args, **args_dict):
+            pass
+
+        def warning(self, *args, **args_dict):
+            pass
+
+    class MockMyCli:
+        connect_calls = 0
+        ran_queries = []
+
+        config = {
+            'main': {
+                'use_keyring': 'False',
+                'my_cnf_transition_done': 'True',
+            },
+            'connection': {},
+        }
+
+        def __init__(self, **_args):
+            self.logger = Logger()
+            self.destructive_warning = False
+            self.main_formatter = Formatter()
+            self.redirect_formatter = Formatter()
+            self.ssl_mode = 'auto'
+            self.my_cnf = {'client': {}, 'mysqld': {}}
+            self.default_keepalive_ticks = 0
+            self.config_without_package_defaults = {'connection': {}}
+
+        def connect(self, **_args):
+            MockMyCli.connect_calls += 1
+
+        def run_query(self, query, checkpoint=None, new_line=True):
+            MockMyCli.ran_queries.append(query)
+
+        def run_cli(self):
+            raise AssertionError('should not enter interactive cli')
+
+        def close(self):
+            pass
+
+    import mycli.main
+
+    monkeypatch.setattr(mycli.main, 'MyCli', MockMyCli)
+    return mycli.main, MockMyCli
+
+
+def test_batch_file(monkeypatch):
+    mycli_main, MockMyCli = _noninteractive_mock_mycli(monkeypatch)
+    runner = CliRunner()
+
+    with NamedTemporaryFile(prefix=TEMPFILE_PREFIX, mode='w', delete=False) as batch_file:
+        batch_file.write('select 2;')
+        batch_file.flush()
+
+    try:
+        result = runner.invoke(
+            mycli_main.cli,
+            args=['--batch', batch_file.name],
+        )
+        assert result.exit_code == 0
+        assert MockMyCli.ran_queries == ['select 2;']
+    finally:
+        os.remove(batch_file.name)
+
+
+def test_execute_arg_warns_about_ignoring_stdin(monkeypatch):
+    mycli_main, MockMyCli = _noninteractive_mock_mycli(monkeypatch)
+    runner = CliRunner()
+
+    # the test env should make sure stdin is not a TTY
+    result = runner.invoke(mycli_main.cli, args=['--execute', 'select 1;'])
+
+    # this exit_code is as written currently, but a debatable choice,
+    # since there was a warning
+    assert result.exit_code == 0
+    assert 'Ignoring STDIN' in result.output
+
+
+def test_batch_file_open_error(monkeypatch):
+    mycli_main, MockMyCli = _noninteractive_mock_mycli(monkeypatch)
+    runner = CliRunner()
+
+    result = runner.invoke(mycli_main.cli, args=['--batch', 'definitely_missing_file.sql'])
+
+    assert result.exit_code != 0
+    assert 'Failed to open --batch file' in result.output
+
+
+def test_execute_arg_supersedes_batch_file(monkeypatch):
+    mycli_main, MockMyCli = _noninteractive_mock_mycli(monkeypatch)
+    runner = CliRunner()
+
+    with NamedTemporaryFile(prefix=TEMPFILE_PREFIX, mode='w', delete=False) as batch_file:
+        batch_file.write('select 2;\n')
+        batch_file.flush()
+
+    try:
+        result = runner.invoke(
+            mycli_main.cli,
+            args=['--execute', 'select 1;', '--batch', batch_file.name],
+        )
+        # this exit_code is as written currently, but a debatable choice,
+        # since there was a warning
+        assert result.exit_code == 0
+        assert MockMyCli.ran_queries == ['select 1;']
+    finally:
+        os.remove(batch_file.name)
+
+
 def test_null_string_config(monkeypatch):
     monkeypatch.setattr(MyCli, 'system_config_files', [])
     monkeypatch.setattr(MyCli, 'pwd_config_file', os.devnull)