sqlglot.errors

  1from __future__ import annotations
  2
  3import typing as t
  4from enum import auto
  5from collections.abc import Sequence
  6from sqlglot.helper import AutoName
  7
  8
  9# ANSI escape codes for error formatting
 10ANSI_UNDERLINE = "\033[4m"
 11ANSI_RESET = "\033[0m"
 12ERROR_MESSAGE_CONTEXT_DEFAULT = 100
 13
 14
 15class ErrorLevel(AutoName):
 16    IGNORE = auto()
 17    """Ignore all errors."""
 18
 19    WARN = auto()
 20    """Log all errors."""
 21
 22    RAISE = auto()
 23    """Collect all errors and raise a single exception."""
 24
 25    IMMEDIATE = auto()
 26    """Immediately raise an exception on the first error found."""
 27
 28
 29class SqlglotError(Exception):
 30    pass
 31
 32
 33class UnsupportedError(SqlglotError):
 34    pass
 35
 36
 37class ParseError(SqlglotError):
 38    def __init__(
 39        self,
 40        message: str,
 41        errors: list[dict[str, t.Any]] | None = None,
 42    ):
 43        super().__init__(message)
 44        self.errors = errors or []
 45
 46    @classmethod
 47    def new(
 48        cls,
 49        message: str,
 50        description: str | None = None,
 51        line: int | None = None,
 52        col: int | None = None,
 53        start_context: str | None = None,
 54        highlight: str | None = None,
 55        end_context: str | None = None,
 56        into_expression: str | None = None,
 57    ) -> ParseError:
 58        return cls(
 59            message,
 60            [
 61                {
 62                    "description": description,
 63                    "line": line,
 64                    "col": col,
 65                    "start_context": start_context,
 66                    "highlight": highlight,
 67                    "end_context": end_context,
 68                    "into_expression": into_expression,
 69                }
 70            ],
 71        )
 72
 73
 74class TokenError(SqlglotError):
 75    """Error raised when tokenizing fails.
 76
 77    When available, `start` and `end` are the offsets in the source SQL of the context
 78    snippet quoted in the message, i.e. the snippet is `sql[start:end]`.
 79    """
 80
 81    def __init__(
 82        self,
 83        message: str,
 84        start: int | None = None,
 85        end: int | None = None,
 86    ):
 87        super().__init__(message)
 88        self.start = start
 89        self.end = end
 90
 91
 92class OptimizeError(SqlglotError):
 93    pass
 94
 95
 96class SchemaError(SqlglotError):
 97    pass
 98
 99
100class ExecuteError(SqlglotError):
101    pass
102
103
104def highlight_sql(
105    sql: str,
106    positions: list[tuple[int, int]],
107    context_length: int = ERROR_MESSAGE_CONTEXT_DEFAULT,
108) -> tuple[str, str, str, str]:
109    """
110    Highlight a SQL string using ANSI codes at the given positions.
111
112    Args:
113        sql: The complete SQL string.
114        positions: List of (start, end) tuples where both start and end are inclusive 0-based
115            indexes. For example, to highlight "foo" in "SELECT foo", use (7, 9).
116            The positions will be sorted and de-duplicated if they overlap.
117        context_length: Number of characters to show before the first highlight and after
118            the last highlight.
119
120    Returns:
121        A tuple of (formatted_sql, start_context, highlight, end_context) where:
122        - formatted_sql: The SQL with ANSI underline codes applied to highlighted sections
123        - start_context: Plain text before the first highlight
124        - highlight: Plain text from the first highlight start to the last highlight end,
125            including any non-highlighted text in between (no ANSI)
126        - end_context: Plain text after the last highlight
127
128    Note:
129        If positions is empty, raises a ValueError.
130    """
131    if not positions:
132        raise ValueError("positions must contain at least one (start, end) tuple")
133
134    start_context = ""
135    end_context = ""
136    first_highlight_start = 0
137    formatted_parts = []
138    previous_part_end = 0
139    sorted_positions = sorted(positions, key=lambda pos: pos[0])
140
141    if sorted_positions[0][0] > 0:
142        first_highlight_start = sorted_positions[0][0]
143        start_context = sql[max(0, first_highlight_start - context_length) : first_highlight_start]
144        formatted_parts.append(start_context)
145        previous_part_end = first_highlight_start
146
147    for start, end in sorted_positions:
148        highlight_start = max(start, previous_part_end)
149        highlight_end = end + 1
150        if highlight_start >= highlight_end:
151            continue  # Skip invalid or overlapping highlights
152        if highlight_start > previous_part_end:
153            formatted_parts.append(sql[previous_part_end:highlight_start])
154        formatted_parts.append(f"{ANSI_UNDERLINE}{sql[highlight_start:highlight_end]}{ANSI_RESET}")
155        previous_part_end = highlight_end
156
157    if previous_part_end < len(sql):
158        end_context = sql[previous_part_end : previous_part_end + context_length]
159        formatted_parts.append(end_context)
160
161    formatted_sql = "".join(formatted_parts)
162    highlight = sql[first_highlight_start:previous_part_end]
163
164    return formatted_sql, start_context, highlight, end_context
165
166
167def concat_messages(errors: Sequence[t.Any], maximum: int) -> str:
168    msg = [str(e) for e in errors[:maximum]]
169    remaining = len(errors) - maximum
170    if remaining > 0:
171        msg.append(f"... and {remaining} more")
172    return "\n\n".join(msg)
173
174
175def merge_errors(errors: Sequence[ParseError]) -> list[dict[str, t.Any]]:
176    return [e_dict for error in errors for e_dict in error.errors]