Source code for dagster_dbt.errors
import warnings
from abc import ABC
from typing import Any, Mapping, Optional, Sequence
from dagster import Failure, MetadataValue
from dagster import _check as check
[docs]class DagsterDbtError(Failure, ABC):
"""The base exception of the ``dagster-dbt`` library."""
[docs]class DagsterDbtCliUnexpectedOutputError(DagsterDbtError):
"""Represents an error when parsing the output of a dbt CLI command."""
invalid_line_nos: Sequence[int]
def __init__(self, invalid_line_nos: Sequence[int]):
check.sequence_param(invalid_line_nos, "invalid_line_nos", int)
line_nos_str = ", ".join(map(str, invalid_line_nos))
description = f"dbt CLI emitted unexpected output on lines {line_nos_str}"
metadata = {
"Invalid CLI Output Line Numbers": MetadataValue.json({"line_nos": invalid_line_nos})
}
super().__init__(description, metadata=metadata)
self.invalid_line_nos = invalid_line_nos
[docs]class DagsterDbtCliRuntimeError(DagsterDbtError, ABC):
"""Represents an error while executing a dbt CLI command."""
def __init__(
self,
description: str,
logs: Optional[Sequence[Mapping[str, Any]]] = None,
raw_output: Optional[str] = None,
messages: Optional[Sequence[str]] = None,
):
if logs is not None:
warnings.warn(
"`logs` is a deprecated argument to DagsterDbtCliRuntimeError and will be discarded"
)
if raw_output is not None:
warnings.warn(
"`raw_output` is a deprecated argument to DagsterDbtCliRuntimeError and will be discarded"
)
metadata = {"Parsed CLI Messages": "\n".join(messages or [])}
super().__init__(description, metadata=metadata)
[docs]class DagsterDbtCliHandledRuntimeError(DagsterDbtCliRuntimeError):
"""Represents a model error reported by the dbt CLI at runtime (return code 1)."""
def __init__(
self,
logs: Optional[Sequence[Mapping[str, Any]]] = None,
raw_output: Optional[str] = None,
messages: Optional[Sequence[str]] = None,
):
super().__init__("Handled error in the dbt CLI (return code 1)", logs, raw_output, messages)
[docs]class DagsterDbtCliFatalRuntimeError(DagsterDbtCliRuntimeError):
"""Represents a fatal error in the dbt CLI (return code 2)."""
def __init__(
self,
logs: Optional[Sequence[Mapping[str, Any]]] = None,
raw_output: Optional[str] = None,
messages: Optional[Sequence[str]] = None,
):
super().__init__(
"Fatal error in the dbt CLI (return code 2): " + " ".join(messages or []),
logs,
raw_output,
messages,
)
[docs]class DagsterDbtRpcUnexpectedPollOutputError(DagsterDbtError):
"""Represents an unexpected response when polling the dbt RPC server."""
[docs]class DagsterDbtCliOutputsNotFoundError(DagsterDbtError):
"""Represents a problem in finding the ``target/run_results.json`` artifact when executing a dbt
CLI command.
For more details on ``target/run_results.json``, see
https://docs.getdbt.com/reference/dbt-artifacts#run_resultsjson.
"""
def __init__(self, path: str):
super().__init__("Expected to find file at path {}".format(path))