sqlmesh.utils.errors

  1from __future__ import annotations
  2
  3import typing as t
  4from enum import auto
  5from pathlib import Path
  6
  7from sqlglot import exp
  8from sqlglot.helper import AutoName
  9
 10if t.TYPE_CHECKING:
 11    from sqlmesh.core.model import Model
 12
 13
 14class ErrorLevel(AutoName):
 15    IGNORE = auto()
 16    WARN = auto()
 17    RAISE = auto()
 18
 19
 20class SQLMeshError(Exception):
 21    pass
 22
 23
 24class ConfigError(SQLMeshError):
 25    pass
 26
 27
 28class MissingDependencyError(SQLMeshError):
 29    """Local environment is missing a required dependency for the given operation"""
 30
 31
 32class MacroEvalError(SQLMeshError):
 33    pass
 34
 35
 36class PlanError(SQLMeshError):
 37    pass
 38
 39
 40class NoChangesPlanError(PlanError):
 41    pass
 42
 43
 44class UncategorizedPlanError(PlanError):
 45    pass
 46
 47
 48class MissingContextException(Exception):
 49    pass
 50
 51
 52class SnapshotVersionError(SQLMeshError):
 53    pass
 54
 55
 56class MagicError(SQLMeshError):
 57    pass
 58
 59
 60class AuditConfigError(ConfigError):
 61    pass
 62
 63
 64class AuditError(SQLMeshError):
 65    def __init__(
 66        self,
 67        audit_name: str,
 68        count: int,
 69        query: exp.Query,
 70        model: t.Optional[Model] = None,
 71        # the dialect of the engine adapter that evaluated the audit query
 72        adapter_dialect: t.Optional[str] = None,
 73    ) -> None:
 74        self.audit_name = audit_name
 75        self.model = model
 76        self.count = count
 77        self.query = query
 78        self.adapter_dialect = adapter_dialect
 79
 80    def __str__(self) -> str:
 81        model_str = f" for model '{self.model_name}'" if self.model_name else ""
 82        return f"Audit '{self.audit_name}'{model_str} failed.\nGot {self.count} results, expected 0.\n{self.sql()}"
 83
 84    @property
 85    def model_name(self) -> t.Optional[str]:
 86        return self.model.name if self.model else None
 87
 88    def sql(self, dialect: t.Optional[str] = None, **opts: t.Any) -> str:
 89        """
 90        Returns the rendered audit query that failed.
 91
 92        Args:
 93            dialect: the dialect of the output SQL string, by default,
 94                     this will use the dialect of the engine adapter that ran the query.
 95            opts: other `sqlglot.generator.Generator` options.
 96
 97        Returns:
 98            The SQL string.
 99        """
100        return self.query.sql(dialect=dialect or self.adapter_dialect, **opts)
101
102
103class TestError(SQLMeshError):
104    pass
105
106
107class NotificationTargetError(SQLMeshError):
108    pass
109
110
111class ApiError(SQLMeshError):
112    pass
113
114
115class ApiClientError(ApiError):
116    pass
117
118
119class ApiServerError(ApiError):
120    pass
121
122
123class NotFoundError(ApiClientError):
124    pass
125
126
127class CICDBotError(SQLMeshError):
128    pass
129
130
131class ParsetimeAdapterCallError(SQLMeshError):
132    pass
133
134
135class EngineAdapterError(SQLMeshError):
136    pass
137
138
139class UnsupportedCatalogOperationError(EngineAdapterError):
140    pass
141
142
143class CircuitBreakerError(SQLMeshError):
144    def __init__(self) -> None:
145        super().__init__("Circuit breaker has been triggered.")
146
147
148def raise_config_error(
149    msg: str,
150    location: t.Optional[str | Path] = None,
151    error_type: t.Type[ConfigError] = ConfigError,
152) -> None:
153    if location:
154        raise error_type(f"{msg} at '{location}'")
155    raise error_type(msg)