Source code for dagster_dbt.errors

import warnings
from abc import ABC
from typing import Any, Mapping, Optional, Sequence

from dagster import (
    DagsterInvariantViolationError,
    Failure,
    MetadataValue,
    _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 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__(f"Expected to find file at path {path}")


class DagsterDbtCloudJobInvariantViolationError(DagsterDbtError, DagsterInvariantViolationError):
    """Represents an error when a dbt Cloud job is not supported by the ``dagster-dbt`` library."""


class DagsterDbtProjectNotFoundError(DagsterDbtError):
    """Error when the specified project directory can not be found."""


class DagsterDbtManifestNotFoundError(DagsterDbtError):
    """Error when we expect manifest.json to generated already but it is absent."""