Add support for -f/--file option to execute SQL from files

changelog.rst

@@ -34,6 +34,9 @@ Features:
     * Support dsn specific init-command in the config file
 * Add suggestion when setting the search_path
 * Allow per dsn_alias ssh tunnel selection
+* Add support for executing SQL commands from file and exit.
+    * Command line option `-f` or `--file`.
+    * Multiple files can be specified.
 
 Internal:
 ---------

pgcli/main.py

@@ -959,6 +959,32 @@ def _check_ongoing_transaction_and_allow_quitting(self):
     def run_cli(self):
         logger = self.logger
 
+        # Handle file mode (-f flag) - similar to psql behavior
+        # Multiple -f options are executed sequentially
+        if hasattr(self, 'input_files') and self.input_files:
+            try:
+                for input_file in self.input_files:
+                    logger.debug("Reading commands from file: %s", input_file)
+                    with open(input_file, 'r', encoding='utf-8') as f:
+                        file_content = f.read()
+
+                    # Execute the entire file content as a single command
+                    # This matches psql behavior where the file is treated as one unit
+                    if file_content.strip():
+                        logger.debug("Executing commands from file: %s", input_file)
+                        self.handle_watch_command(file_content)
+
+            except PgCliQuitError:
+                # Normal exit from quit command
+                sys.exit(0)
+            except Exception as e:
+                logger.error("Error executing command: %s", e)
+                logger.error("traceback: %r", traceback.format_exc())
+                click.secho(str(e), err=True, fg="red")
+                sys.exit(1)
+            # Exit successfully after executing all commands
+            sys.exit(0)
+
         history_file = self.config["main"]["history_file"]
         if history_file == "default":
             history_file = config_location() + "history"
@@ -1340,7 +1366,8 @@ def is_too_tall(self, lines):
         return len(lines) >= (self.prompt_app.output.get_size().rows - 4)
 
     def echo_via_pager(self, text, color=None):
-        if self.pgspecial.pager_config == PAGER_OFF or self.watch_command:
+        # Disable pager for -f/--file mode and \watch command
+        if self.pgspecial.pager_config == PAGER_OFF or self.watch_command or (hasattr(self, 'input_files') and self.input_files):
             click.echo(text, color=color)
         elif self.pgspecial.pager_config == PAGER_LONG_OUTPUT and self.table_format != "csv":
             lines = text.split("\n")
@@ -1488,6 +1515,14 @@ def echo_via_pager(self, text, color=None):
     type=str,
     help="SQL statement to execute after connecting.",
 )
+@click.option(
+    "-f",
+    "--file",
+    "input_files",
+    multiple=True,
+    type=click.Path(exists=True, readable=True, dir_okay=False),
+    help="execute commands from file, then exit. Multiple -f options are allowed.",
+)
 @click.argument("dbname", default=lambda: None, envvar="PGDATABASE", nargs=1)
 @click.argument("username", default=lambda: None, envvar="PGUSER", nargs=1)
 def cli(
@@ -1516,6 +1551,7 @@ def cli(
     ssh_tunnel: str,
     init_command: str,
     log_file: str,
+    input_files: tuple,
 ):
     if version:
         print("Version:", __version__)
@@ -1576,6 +1612,9 @@ def cli(
         log_file=log_file,
     )
 
+    # Store file paths for -f option (can be multiple)
+    pgcli.input_files = input_files if input_files else None
+
     # Choose which ever one has a valid value.
     if dbname_opt and dbname:
         # work as psql: when database is given as option and argument use the argument as user

tests/features/file_option.feature

@@ -0,0 +1,33 @@
+Feature: run the cli with -f/--file option,
+  execute commands from file,
+  and exit
+
+  Scenario: run pgcli with -f and a SQL query file
+     When we create a file with "SELECT 1 as test_diego_column"
+      and we run pgcli with -f and the file
+     then we see the query result
+      and pgcli exits successfully
+
+  Scenario: run pgcli with --file and a SQL query file
+     When we create a file with "SELECT 'hello' as greeting"
+      and we run pgcli with --file and the file
+     then we see the query result
+      and pgcli exits successfully
+
+  Scenario: run pgcli with -f and a file with special command
+     When we create a file with "\dt"
+      and we run pgcli with -f and the file
+     then we see the command output
+      and pgcli exits successfully
+
+  Scenario: run pgcli with -f and a file with multiple statements
+     When we create a file with "SELECT 1; SELECT 2"
+      and we run pgcli with -f and the file
+     then we see both query results
+      and pgcli exits successfully
+
+  Scenario: run pgcli with -f and a file with an invalid query
+     When we create a file with "SELECT invalid_column FROM nonexistent_table"
+      and we run pgcli with -f and the file
+     then we see an error message
+      and pgcli exits successfully

tests/features/steps/file_option.py

@@ -0,0 +1,137 @@
+"""
+Steps for testing -f/--file option behavioral tests.
+"""
+
+import subprocess
+import tempfile
+import os
+from behave import when, then
+
+
+@when('we create a file with "{content}"')
+def step_create_file_with_content(context, content):
+    """Create a temporary file with the given content."""
+    # Create a temporary file that will be cleaned up automatically
+    temp_file = tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.sql')
+    temp_file.write(content)
+    temp_file.close()
+    context.temp_file_path = temp_file.name
+
+
+@when('we run pgcli with -f and the file')
+def step_run_pgcli_with_f(context):
+    """Run pgcli with -f flag and the temporary file."""
+    cmd = [
+        "pgcli",
+        "-h",
+        context.conf["host"],
+        "-p",
+        str(context.conf["port"]),
+        "-U",
+        context.conf["user"],
+        "-d",
+        context.conf["dbname"],
+        "-f",
+        context.temp_file_path,
+    ]
+    try:
+        context.cmd_output = subprocess.check_output(cmd, cwd=context.package_root, stderr=subprocess.STDOUT, timeout=5)
+        context.exit_code = 0
+    except subprocess.CalledProcessError as e:
+        context.cmd_output = e.output
+        context.exit_code = e.returncode
+    except subprocess.TimeoutExpired:
+        context.cmd_output = b"Command timed out"
+        context.exit_code = -1
+    finally:
+        # Clean up the temporary file
+        if hasattr(context, 'temp_file_path') and os.path.exists(context.temp_file_path):
+            os.unlink(context.temp_file_path)
+
+
+@when('we run pgcli with --file and the file')
+def step_run_pgcli_with_file(context):
+    """Run pgcli with --file flag and the temporary file."""
+    cmd = [
+        "pgcli",
+        "-h",
+        context.conf["host"],
+        "-p",
+        str(context.conf["port"]),
+        "-U",
+        context.conf["user"],
+        "-d",
+        context.conf["dbname"],
+        "--file",
+        context.temp_file_path,
+    ]
+    try:
+        context.cmd_output = subprocess.check_output(cmd, cwd=context.package_root, stderr=subprocess.STDOUT, timeout=5)
+        context.exit_code = 0
+    except subprocess.CalledProcessError as e:
+        context.cmd_output = e.output
+        context.exit_code = e.returncode
+    except subprocess.TimeoutExpired:
+        context.cmd_output = b"Command timed out"
+        context.exit_code = -1
+    finally:
+        # Clean up the temporary file
+        if hasattr(context, 'temp_file_path') and os.path.exists(context.temp_file_path):
+            os.unlink(context.temp_file_path)
+
+
+@then("we see the query result")
+def step_see_query_result(context):
+    """Verify that the query result is in the output."""
+    output = context.cmd_output.decode('utf-8')
+    # Check for common query result indicators
+    assert any([
+        "SELECT" in output,
+        "test_diego_column" in output,
+        "greeting" in output,
+        "hello" in output,
+        "+-" in output,  # table border
+        "|" in output,  # table column separator
+    ]), f"Expected query result in output, but got: {output}"
+
+
+@then("we see both query results")
+def step_see_both_query_results(context):
+    """Verify that both query results are in the output."""
+    output = context.cmd_output.decode('utf-8')
+    # Should contain output from both SELECT statements
+    assert "SELECT" in output, f"Expected SELECT in output, but got: {output}"
+    # The output should have multiple result sets
+    assert output.count("SELECT") >= 2, f"Expected at least 2 SELECT results, but got: {output}"
+
+
+@then("we see the command output")
+def step_see_command_output(context):
+    """Verify that the special command output is present."""
+    output = context.cmd_output.decode('utf-8')
+    # `\dt` renders its column headers whether or not any tables exist, so the
+    # headers are what tells us the special command actually ran and produced
+    # its listing (rather than erroring out).
+    for header in ("Schema", "Name", "Type", "Owner"):
+        assert header in output, f"Expected {header!r} in \\dt output, but got: {output}"
+    assert context.exit_code == 0, f"Expected exit code 0, but got: {context.exit_code}"
+
+
+@then("we see an error message")
+def step_see_error_message(context):
+    """Verify that an error message is in the output."""
+    output = context.cmd_output.decode('utf-8')
+    assert any([
+        "does not exist" in output,
+        "error" in output.lower(),
+        "ERROR" in output,
+    ]), f"Expected error message in output, but got: {output}"
+
+
+@then("pgcli exits successfully")
+def step_pgcli_exits_successfully(context):
+    """Verify that pgcli exited with code 0."""
+    assert context.exit_code == 0, f"Expected exit code 0, but got: {context.exit_code}"
+    # Clean up
+    context.cmd_output = None
+    context.exit_code = None