sqlglot.errors

 1from __future__ import annotations
 2
 3import typing as t
 4from enum import auto
 5
 6from sqlglot.helper import AutoName
 7
 8
 9class ErrorLevel(AutoName):
10    IGNORE = auto()
11    """Ignore all errors."""
12
13    WARN = auto()
14    """Log all errors."""
15
16    RAISE = auto()
17    """Collect all errors and raise a single exception."""
18
19    IMMEDIATE = auto()
20    """Immediately raise an exception on the first error found."""
21
22
23class SqlglotError(Exception):
24    pass
25
26
27class UnsupportedError(SqlglotError):
28    pass
29
30
31class ParseError(SqlglotError):
32    def __init__(
33        self,
34        message: str,
35        errors: t.Optional[t.List[t.Dict[str, t.Any]]] = None,
36    ):
37        super().__init__(message)
38        self.errors = errors or []
39
40    @classmethod
41    def new(
42        cls,
43        message: str,
44        description: t.Optional[str] = None,
45        line: t.Optional[int] = None,
46        col: t.Optional[int] = None,
47        start_context: t.Optional[str] = None,
48        highlight: t.Optional[str] = None,
49        end_context: t.Optional[str] = None,
50        into_expression: t.Optional[str] = None,
51    ) -> ParseError:
52        return cls(
53            message,
54            [
55                {
56                    "description": description,
57                    "line": line,
58                    "col": col,
59                    "start_context": start_context,
60                    "highlight": highlight,
61                    "end_context": end_context,
62                    "into_expression": into_expression,
63                }
64            ],
65        )
66
67
68class TokenError(SqlglotError):
69    pass
70
71
72class OptimizeError(SqlglotError):
73    pass
74
75
76class SchemaError(SqlglotError):
77    pass
78
79
80class ExecuteError(SqlglotError):
81    pass
82
83
84def concat_messages(errors: t.Sequence[t.Any], maximum: int) -> str:
85    msg = [str(e) for e in errors[:maximum]]
86    remaining = len(errors) - maximum
87    if remaining > 0:
88        msg.append(f"... and {remaining} more")
89    return "\n\n".join(msg)
90
91
92def merge_errors(errors: t.Sequence[ParseError]) -> t.List[t.Dict[str, t.Any]]:
93    return [e_dict for error in errors for e_dict in error.errors]