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