feat: add functionality to run single SQL statement at cursor (g+s) (#89)f

* feat: add functionality to run single SQL statement at cursor (g+s)

* fix: support blank-line separated statements in g+s

_get_statement_ranges() now handles blank-line separated statements,
matching the behavior of split_statements(). Previously, g+s would
run the entire buffer when statements were separated by blank lines
instead of semicolons.

* refactor: extract shared SQL char iterator to reduce duplication

The quote-tracking state machine was duplicated 5 times. Now uses
a single _iter_sql_chars() generator that handles escape sequences
and string literal detection, reducing the file by ~100 lines.

---------

Co-authored-by: Peter Adams <18162810+Maxteabag@users.noreply.github.com>

Author: LVG77

sqlit/core/keymap.py

@@ -274,6 +274,7 @@ def _build_leader_commands(self) -> list[LeaderCommandDef]:
             LeaderCommandDef("E", "WORD_end_back", "End of prev WORD", "Go to", menu="g"),
             LeaderCommandDef("c", "comment", "Toggle comment...", "Toggle", menu="g"),
             LeaderCommandDef("r", "execute_query", "Run query", "Execute", menu="g"),
+            LeaderCommandDef("s", "execute_single_statement", "Run statement at cursor", "Execute", menu="g"),
             LeaderCommandDef("t", "execute_query_atomic", "Run as transaction", "Execute", menu="g"),
             # gc comment menu (vim-style)
             LeaderCommandDef("c", "line", "Toggle line comment", "Comment", menu="gc"),

sqlit/domains/query/app/multi_statement.py

@@ -10,94 +10,82 @@
 
 import re
 from dataclasses import dataclass, field
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Iterator
 
 if TYPE_CHECKING:
     from .query_service import NonQueryResult, QueryResult
 
 
-def _has_semicolon_outside_strings(sql: str) -> bool:
-    """Check if SQL has semicolons outside of string literals."""
+def _iter_sql_chars(sql: str) -> Iterator[tuple[int, str, bool]]:
+    """Iterate through SQL characters, tracking string literal context.
+
+    Handles escape sequences (backslash) and SQL-style doubled quotes.
+
+    Yields:
+        (index, char, outside_string) tuples where outside_string is True
+        when the character is not inside a string literal.
+    """
     in_single_quote = False
     in_double_quote = False
     i = 0
 
     while i < len(sql):
         char = sql[i]
 
-        # Handle escape sequences
+        # Handle escape sequences in strings
         if i + 1 < len(sql) and char == "\\" and (in_single_quote or in_double_quote):
+            yield (i, char, False)
+            yield (i + 1, sql[i + 1], False)
             i += 2
             continue
 
-        # Handle doubled quotes
+        # Handle doubled quotes (SQL escape for quotes)
         if char == "'" and i + 1 < len(sql) and sql[i + 1] == "'" and in_single_quote:
+            yield (i, "'", False)
+            yield (i + 1, "'", False)
             i += 2
             continue
         if char == '"' and i + 1 < len(sql) and sql[i + 1] == '"' and in_double_quote:
+            yield (i, '"', False)
+            yield (i + 1, '"', False)
             i += 2
             continue
 
-        # Toggle quote state
+        # Toggle quote state and yield
         if char == "'" and not in_double_quote:
             in_single_quote = not in_single_quote
+            yield (i, char, False)  # Quote char is part of string syntax
         elif char == '"' and not in_single_quote:
             in_double_quote = not in_double_quote
-        elif char == ";" and not in_single_quote and not in_double_quote:
-            return True
+            yield (i, char, False)  # Quote char is part of string syntax
+        else:
+            yield (i, char, not in_single_quote and not in_double_quote)
 
         i += 1
 
+
+def _has_semicolon_outside_strings(sql: str) -> bool:
+    """Check if SQL has semicolons outside of string literals."""
+    for _, char, outside in _iter_sql_chars(sql):
+        if char == ";" and outside:
+            return True
     return False
 
 
 def _split_by_semicolons(sql: str) -> list[str]:
     """Split SQL by semicolons, respecting string literals."""
     statements = []
-    current = []
-    in_single_quote = False
-    in_double_quote = False
-    i = 0
-
-    while i < len(sql):
-        char = sql[i]
-
-        # Handle escape sequences in strings
-        if i + 1 < len(sql) and char == "\\" and (in_single_quote or in_double_quote):
-            current.append(char)
-            current.append(sql[i + 1])
-            i += 2
-            continue
-
-        # Handle doubled quotes (SQL escape for quotes)
-        if char == "'" and i + 1 < len(sql) and sql[i + 1] == "'" and in_single_quote:
-            current.append("''")
-            i += 2
-            continue
+    current: list[str] = []
 
-        if char == '"' and i + 1 < len(sql) and sql[i + 1] == '"' and in_double_quote:
-            current.append('""')
-            i += 2
-            continue
-
-        # Toggle quote state
-        if char == "'" and not in_double_quote:
-            in_single_quote = not in_single_quote
-            current.append(char)
-        elif char == '"' and not in_single_quote:
-            in_double_quote = not in_double_quote
-            current.append(char)
-        elif char == ";" and not in_single_quote and not in_double_quote:
-            # End of statement
+    for _, char, outside in _iter_sql_chars(sql):
+        if char == ";" and outside:
             stmt = "".join(current).strip()
             if stmt:
                 statements.append(stmt)
             current = []
         else:
             current.append(char)
 
-        i += 1
-
     # Don't forget the last statement (may not end with semicolon)
     stmt = "".join(current).strip()
     if stmt:
@@ -113,51 +101,20 @@ def _split_by_blank_lines(sql: str) -> list[str]:
     This is triggered when there are no semicolons in the query.
     """
     statements = []
-    current = []
-    in_single_quote = False
-    in_double_quote = False
-    i = 0
+    current: list[str] = []
     line_start = 0
     prev_line_empty = False
 
-    while i < len(sql):
-        char = sql[i]
-
-        # Handle escape sequences in strings
-        if i + 1 < len(sql) and char == "\\" and (in_single_quote or in_double_quote):
-            current.append(char)
-            current.append(sql[i + 1])
-            i += 2
-            continue
-
-        # Handle doubled quotes (SQL escape for quotes)
-        if char == "'" and i + 1 < len(sql) and sql[i + 1] == "'" and in_single_quote:
-            current.append("''")
-            i += 2
-            continue
-
-        if char == '"' and i + 1 < len(sql) and sql[i + 1] == '"' and in_double_quote:
-            current.append('""')
-            i += 2
-            continue
-
-        # Toggle quote state
-        if char == "'" and not in_double_quote:
-            in_single_quote = not in_single_quote
-            current.append(char)
-        elif char == '"' and not in_single_quote:
-            in_double_quote = not in_double_quote
-            current.append(char)
-        elif char == "\n" and not in_single_quote and not in_double_quote:
-            # Check if this line (from line_start to i) is empty/whitespace
-            line_content = sql[line_start:i]
+    for idx, char, outside in _iter_sql_chars(sql):
+        if char == "\n" and outside:
+            line_content = sql[line_start:idx]
             current_line_empty = not line_content.strip()
 
             if current_line_empty and prev_line_empty:
-                # We have a blank line separator - don't add more newlines
+                # Consecutive blank lines, skip
                 pass
             elif current_line_empty and current:
-                # This is a blank line after content - split here
+                # Blank line after content - split here
                 stmt = "".join(current).strip()
                 if stmt:
                     statements.append(stmt)
@@ -167,14 +124,12 @@ def _split_by_blank_lines(sql: str) -> list[str]:
                 current.append(char)
 
             prev_line_empty = current_line_empty
-            line_start = i + 1
+            line_start = idx + 1
         else:
             current.append(char)
-            if char not in " \t":
+            if char not in " \t\n":
                 prev_line_empty = False
 
-        i += 1
-
     # Don't forget the last statement
     stmt = "".join(current).strip()
     if stmt:
@@ -183,6 +138,125 @@ def _split_by_blank_lines(sql: str) -> list[str]:
     return statements
 
 
+def _append_statement_range(
+    ranges: list[tuple[str, int, int]], sql: str, stmt_start: int, stmt_end: int
+) -> None:
+    """Helper to append a statement range, calculating actual positions."""
+    stmt_full = sql[stmt_start:stmt_end]
+    stmt_text = stmt_full.strip()
+    if stmt_text:
+        actual_start = stmt_start + (len(stmt_full) - len(stmt_full.lstrip()))
+        ranges.append((stmt_text, actual_start, actual_start + len(stmt_text)))
+
+
+def _get_statement_ranges(sql: str) -> list[tuple[str, int, int]]:
+    """Get statements with their character ranges in the original SQL.
+
+    Splitting strategy (matches split_statements):
+    1. If query contains semicolons (outside strings) → split by semicolons
+    2. If no semicolons but has blank lines → split by blank lines
+    3. Otherwise → return as single statement
+
+    Returns:
+        List of (statement_text, start_offset, end_offset) tuples.
+        Offsets are 0-based character positions in the original SQL string.
+    """
+    if not sql or not sql.strip():
+        return []
+
+    ranges: list[tuple[str, int, int]] = []
+
+    # Strategy 1: If semicolons exist, use semicolon splitting with tracking
+    if _has_semicolon_outside_strings(sql):
+        stmt_start = 0
+
+        for idx, char, outside in _iter_sql_chars(sql):
+            if char == ";" and outside:
+                _append_statement_range(ranges, sql, stmt_start, idx)
+                stmt_start = idx + 1
+
+        _append_statement_range(ranges, sql, stmt_start, len(sql))
+        return ranges
+
+    # Strategy 2: If blank lines exist, use blank line splitting with tracking
+    if re.search(r"\n\s*\n", sql):
+        stmt_start = 0
+        line_start = 0
+        prev_line_empty = False
+
+        for idx, char, outside in _iter_sql_chars(sql):
+            if char == "\n" and outside:
+                line_content = sql[line_start:idx]
+                current_line_empty = not line_content.strip()
+
+                if current_line_empty and prev_line_empty:
+                    # Consecutive blank lines, skip
+                    pass
+                elif current_line_empty:
+                    # Blank line after content - this is a statement boundary
+                    _append_statement_range(ranges, sql, stmt_start, idx)
+                    stmt_start = idx + 1
+
+                prev_line_empty = current_line_empty
+                line_start = idx + 1
+            elif char not in " \t\n":
+                prev_line_empty = False
+
+        _append_statement_range(ranges, sql, stmt_start, len(sql))
+        return ranges
+
+    # Strategy 3: Single statement
+    stripped = sql.strip()
+    if stripped:
+        start_offset = len(sql) - len(sql.lstrip())
+        return [(stripped, start_offset, len(sql))]
+
+    return []
+
+
+def find_statement_at_cursor(sql: str, row: int, col: int) -> tuple[str, int, int] | None:
+    """Find the SQL statement containing the cursor position.
+
+    Args:
+        sql: Full SQL text (may contain multiple statements).
+        row: Cursor row (0-based line number).
+        col: Cursor column (0-based character position within the line).
+
+    Returns:
+        Tuple of (statement_text, start_char_offset, end_char_offset) or None if not found.
+    """
+    if not sql:
+        return None
+
+    # Convert (row, col) to absolute character offset
+    lines = sql.split("\n")
+    if row >= len(lines):
+        # Cursor is past end of text, use last position
+        cursor_offset = len(sql)
+    else:
+        # Sum lengths of all previous lines plus newline characters
+        cursor_offset = sum(len(lines[i]) + 1 for i in range(row)) + col
+
+    ranges = _get_statement_ranges(sql)
+
+    if not ranges:
+        return None
+
+    # Find the statement containing the cursor
+    for stmt_text, start, end in ranges:
+        if start <= cursor_offset <= end:
+            return (stmt_text, start, end)
+
+    # If cursor is between statements or at the very end,
+    # return the nearest preceding statement
+    for stmt_text, start, end in reversed(ranges):
+        if cursor_offset >= start:
+            return (stmt_text, start, end)
+
+    # Fallback to first statement
+    return ranges[0] if ranges else None
+
+
 def split_statements(sql: str) -> list[str]:
     """Split SQL into individual statements.
 

sqlit/domains/query/state/query_normal.py

@@ -15,6 +15,7 @@ class QueryNormalModeState(State):
     def _setup_actions(self) -> None:
         self.allows("enter_insert_mode", label="Insert Mode", help="Enter INSERT mode")
         self.allows("execute_query", label="Execute", help="Execute query")
+        self.allows("execute_single_statement", label="Run Statement", help="Execute statement at cursor")
         self.allows("delete_leader_key", label="Delete", help="Delete (menu)")
         self.allows("yank_leader_key", label="Copy", help="Copy (menu)")
         self.allows("change_leader_key", label="Change", help="Change (menu)")

sqlit/domains/query/ui/mixins/query_editing_cursor.py

@@ -40,6 +40,10 @@ def action_g_WORD_end_back(self: QueryMixinHost) -> None:
         self._clear_leader_pending()
         self._move_with_motion("gE")
 
+    def action_g_execute_single_statement(self: QueryMixinHost) -> None:
+        """Execute single statement at cursor via g menu (gs)."""
+        self._clear_leader_pending()
+        self.action_execute_single_statement()
 
     def action_cursor_left(self: QueryMixinHost) -> None:
         """Move cursor left (h in normal mode)."""
@@ -85,7 +89,6 @@ def action_cursor_WORD_back(self: QueryMixinHost) -> None:
         """Move cursor to previous WORD (B)."""
         self._move_with_motion("B")
 
-
     def action_cursor_line_start(self: QueryMixinHost) -> None:
         """Move cursor to start of line (0)."""
         self._move_with_motion("0")