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    from sqlmesh.core.schema_diff import TableAlterOperation
 15
 16
 17class ErrorLevel(AutoName):
 18    IGNORE = auto()
 19    WARN = auto()
 20    RAISE = auto()
 21
 22
 23class SQLMeshError(Exception):
 24    pass
 25
 26
 27class ConfigError(SQLMeshError):
 28    location: t.Optional[Path] = None
 29
 30    def __init__(self, message: str | Exception, location: t.Optional[Path] = None) -> None:
 31        super().__init__(message)
 32        if location:
 33            self.location = Path(location) if isinstance(location, str) else location
 34
 35
 36class BaseMissingReferenceError(ConfigError):
 37    def __init__(self, ref: str) -> None:
 38        self.ref = ref
 39
 40
 41class MissingModelError(BaseMissingReferenceError):
 42    """Raised when a model that is referenced is missing."""
 43
 44
 45class MissingSourceError(BaseMissingReferenceError):
 46    """Raised when a source that is referenced is missing."""
 47
 48
 49class MissingDependencyError(SQLMeshError):
 50    """Local environment is missing a required dependency for the given operation"""
 51
 52
 53class MacroEvalError(SQLMeshError):
 54    pass
 55
 56
 57class PlanError(SQLMeshError):
 58    pass
 59
 60
 61class NoChangesPlanError(PlanError):
 62    pass
 63
 64
 65class UncategorizedPlanError(PlanError):
 66    pass
 67
 68
 69class ConflictingPlanError(PlanError):
 70    pass
 71
 72
 73class MissingContextException(Exception):
 74    pass
 75
 76
 77class SnapshotVersionError(SQLMeshError):
 78    pass
 79
 80
 81class MagicError(SQLMeshError):
 82    pass
 83
 84
 85class AuditConfigError(ConfigError):
 86    pass
 87
 88
 89class StateMigrationError(SQLMeshError):
 90    pass
 91
 92
 93class AuditError(SQLMeshError):
 94    def __init__(
 95        self,
 96        audit_name: str,
 97        audit_args: t.Dict[t.Any, t.Any],
 98        count: int,
 99        query: exp.Query,
100        model: t.Optional[Model] = None,
101        # the dialect of the engine adapter that evaluated the audit query
102        adapter_dialect: t.Optional[str] = None,
103    ) -> None:
104        self.audit_name = audit_name
105        self.audit_args = audit_args
106        self.model = model
107        self.count = count
108        self.query = query
109        self.adapter_dialect = adapter_dialect
110
111        super().__init__(
112            f"'{self.audit_name}' audit error: {self.count} {'row' if self.count == 1 else 'rows'} failed"
113        )
114
115    @property
116    def model_name(self) -> t.Optional[str]:
117        return self.model.name if self.model else None
118
119    def sql(self, dialect: t.Optional[str] = None, **opts: t.Any) -> str:
120        """
121        Returns the rendered audit query that failed.
122
123        Args:
124            dialect: the dialect of the output SQL string, by default,
125                     this will use the dialect of the engine adapter that ran the query.
126            opts: other `sqlglot.generator.Generator` options.
127
128        Returns:
129            The SQL string.
130        """
131        return self.query.sql(dialect=dialect or self.adapter_dialect, **opts)
132
133
134class NodeAuditsErrors(SQLMeshError):
135    def __init__(self, errors: t.List[AuditError]) -> None:
136        self.errors = errors
137
138        super().__init__(f"Audits failed: {', '.join([e.audit_name for e in errors])}")
139
140
141class TestError(SQLMeshError):
142    __test__ = False  # prevent pytest trying to collect this as a test class
143    pass
144
145
146class DestructiveChangeError(SQLMeshError):
147    pass
148
149
150class AdditiveChangeError(SQLMeshError):
151    pass
152
153
154class MigrationNotSupportedError(SQLMeshError):
155    pass
156
157
158class NotificationTargetError(SQLMeshError):
159    pass
160
161
162class ApiError(SQLMeshError):
163    def __init__(self, message: str, code: int) -> None:
164        super().__init__(message)
165        self.code = code
166
167
168class ApiClientError(ApiError):
169    pass
170
171
172class ApiServerError(ApiError):
173    pass
174
175
176class NotFoundError(ApiClientError):
177    def __init__(self, message: str) -> None:
178        super().__init__(message, 404)
179
180
181class CICDBotError(SQLMeshError):
182    pass
183
184
185class ParsetimeAdapterCallError(SQLMeshError):
186    pass
187
188
189class EngineAdapterError(SQLMeshError):
190    pass
191
192
193class UnsupportedCatalogOperationError(EngineAdapterError):
194    pass
195
196
197class CircuitBreakerError(SQLMeshError):
198    def __init__(self) -> None:
199        super().__init__("Circuit breaker triggered.")
200
201
202class PythonModelEvalError(SQLMeshError):
203    pass
204
205
206class MissingDefaultCatalogError(SQLMeshError):
207    pass
208
209
210class LinterError(SQLMeshError):
211    pass
212
213
214class SignalEvalError(SQLMeshError):
215    """Errors when evaluating a signal that is because of a user mistake and not a SQLMesh bug."""
216
217    pass
218
219
220def raise_config_error(
221    msg: str,
222    location: t.Optional[Path] = None,
223    error_type: t.Type[ConfigError] = ConfigError,
224) -> None:
225    if location:
226        raise error_type(f"{msg} at '{location}'", location)
227    raise error_type(msg, location=location)
228
229
230def raise_for_status(response: Response) -> None:
231    if response.status_code == 404:
232        raise NotFoundError(response.text)
233    if 400 <= response.status_code < 500:
234        raise ApiClientError(response.text, response.status_code)
235    if 500 <= response.status_code < 600:
236        raise ApiServerError(response.text, response.status_code)
237
238
239def _format_schema_change_msg(
240    snapshot_name: str,
241    is_destructive: bool,
242    alter_operations: t.List[TableAlterOperation],
243    dialect: str,
244    error: bool = True,
245) -> str:
246    """
247    Common function to format schema change messages.
248
249    Args:
250        snapshot_name: Name of the model/snapshot
251        is_destructive: if change is destructive else it would be additive
252        alter_operations: List of table alter operations
253        dialect: SQL dialect for formatting
254        error: Whether this is an error or warning
255    """
256    from sqlmesh.core.schema_diff import get_dropped_column_names, get_additive_column_names
257
258    change_type = "destructive" if is_destructive else "additive"
259    setting_name = "on_destructive_change" if is_destructive else "on_additive_change"
260    action_verb = "drops" if is_destructive else "adds"
261    cli_flag = "--allow-destructive-model" if is_destructive else "--allow-additive-model"
262
263    column_names = (
264        get_dropped_column_names(alter_operations)
265        if is_destructive
266        else get_additive_column_names(alter_operations)
267    )
268    column_str = "', '".join(column_names)
269    column_msg = (
270        f" that {action_verb} column{'s' if column_names and len(column_names) > 1 else ''} '{column_str}'"
271        if column_str
272        else ""
273    )
274
275    # Format ALTER expressions
276    alter_expr_msg = "\n\nSchema changes:\n  " + "\n  ".join(
277        [alter.expression.sql(dialect) for alter in alter_operations]
278    )
279
280    # Main warning message
281    warning_msg = (
282        f"Plan requires {change_type} change to forward-only model '{snapshot_name}'s schema"
283    )
284
285    if error:
286        permissive_values = "`warn`, `allow`, or `ignore`"
287        cli_part = f" or include the model in the plan's `{cli_flag}` option"
288        err_msg = f"\n\nTo allow the {change_type} change, set the model's `{setting_name}` setting to {permissive_values}{cli_part}.\n"
289    else:
290        err_msg = ""
291
292    return f"\n{warning_msg}{column_msg}.{alter_expr_msg}{err_msg}"
293
294
295def format_destructive_change_msg(
296    snapshot_name: str,
297    alter_expressions: t.List[TableAlterOperation],
298    dialect: str,
299    error: bool = True,
300) -> str:
301    return _format_schema_change_msg(
302        snapshot_name=snapshot_name,
303        is_destructive=True,
304        alter_operations=alter_expressions,
305        dialect=dialect,
306        error=error,
307    )
308
309
310def format_additive_change_msg(
311    snapshot_name: str,
312    alter_operations: t.List[TableAlterOperation],
313    dialect: str,
314    error: bool = True,
315) -> str:
316    return _format_schema_change_msg(
317        snapshot_name=snapshot_name,
318        is_destructive=False,
319        alter_operations=alter_operations,
320        dialect=dialect,
321        error=error,
322    )