sqlglot.errors

  1from __future__ import annotations
  2
  3import typing as t
  4from enum import auto
  5
  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: t.Optional[t.List[t.Dict[str, t.Any]]] = None,
 42    ):
 43        super().__init__(message)
 44        self.errors = errors or []
 45
 46    @classmethod
 47    def new(
 48        cls,
 49        message: str,
 50        description: t.Optional[str] = None,
 51        line: t.Optional[int] = None,
 52        col: t.Optional[int] = None,
 53        start_context: t.Optional[str] = None,
 54        highlight: t.Optional[str] = None,
 55        end_context: t.Optional[str] = None,
 56        into_expression: t.Optional[str] = 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    pass
 76
 77
 78class OptimizeError(SqlglotError):
 79    pass
 80
 81
 82class SchemaError(SqlglotError):
 83    pass
 84
 85
 86class ExecuteError(SqlglotError):
 87    pass
 88
 89
 90def highlight_sql(
 91    sql: str,
 92    positions: t.List[t.Tuple[int, int]],
 93    context_length: int = ERROR_MESSAGE_CONTEXT_DEFAULT,
 94) -> t.Tuple[str, str, str, str]:
 95    """
 96    Highlight a SQL string using ANSI codes at the given positions.
 97
 98    Args:
 99        sql: The complete SQL string.
100        positions: List of (start, end) tuples where both start and end are inclusive 0-based
101            indexes. For example, to highlight "foo" in "SELECT foo", use (7, 9).
102            The positions will be sorted and de-duplicated if they overlap.
103        context_length: Number of characters to show before the first highlight and after
104            the last highlight.
105
106    Returns:
107        A tuple of (formatted_sql, start_context, highlight, end_context) where:
108        - formatted_sql: The SQL with ANSI underline codes applied to highlighted sections
109        - start_context: Plain text before the first highlight
110        - highlight: Plain text from the first highlight start to the last highlight end,
111            including any non-highlighted text in between (no ANSI)
112        - end_context: Plain text after the last highlight
113
114    Note:
115        If positions is empty, raises a ValueError.
116    """
117    if not positions:
118        raise ValueError("positions must contain at least one (start, end) tuple")
119
120    start_context = ""
121    end_context = ""
122    first_highlight_start = 0
123    formatted_parts = []
124    previous_part_end = 0
125    sorted_positions = sorted(positions, key=lambda pos: pos[0])
126
127    if sorted_positions[0][0] > 0:
128        first_highlight_start = sorted_positions[0][0]
129        start_context = sql[max(0, first_highlight_start - context_length) : first_highlight_start]
130        formatted_parts.append(start_context)
131        previous_part_end = first_highlight_start
132
133    for start, end in sorted_positions:
134        highlight_start = max(start, previous_part_end)
135        highlight_end = end + 1
136        if highlight_start >= highlight_end:
137            continue  # Skip invalid or overlapping highlights
138        if highlight_start > previous_part_end:
139            formatted_parts.append(sql[previous_part_end:highlight_start])
140        formatted_parts.append(f"{ANSI_UNDERLINE}{sql[highlight_start:highlight_end]}{ANSI_RESET}")
141        previous_part_end = highlight_end
142
143    if previous_part_end < len(sql):
144        end_context = sql[previous_part_end : previous_part_end + context_length]
145        formatted_parts.append(end_context)
146
147    formatted_sql = "".join(formatted_parts)
148    highlight = sql[first_highlight_start:previous_part_end]
149
150    return formatted_sql, start_context, highlight, end_context
151
152
153def concat_messages(errors: t.Sequence[t.Any], maximum: int) -> str:
154    msg = [str(e) for e in errors[:maximum]]
155    remaining = len(errors) - maximum
156    if remaining > 0:
157        msg.append(f"... and {remaining} more")
158    return "\n\n".join(msg)
159
160
161def merge_errors(errors: t.Sequence[ParseError]) -> t.List[t.Dict[str, t.Any]]:
162    return [e_dict for error in errors for e_dict in error.errors]