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