dbt utils

dbt utilities.

DbtLsException

Bases: Exception

Raised when dbt ls fails.

Source code in src/dbt_score/dbt_utils.py

class DbtLsException(Exception):
    """Raised when dbt ls fails."""

DbtNotInstalledException

Bases: Exception

Raised when trying to run dbt when dbt is not installed.

Source code in src/dbt_score/dbt_utils.py

class DbtNotInstalledException(Exception):
    """Raised when trying to run dbt when dbt is not installed."""

DbtParseException

Bases: Exception

Raised when dbt parse fails.

Source code in src/dbt_score/dbt_utils.py

class DbtParseException(Exception):
    """Raised when dbt parse fails."""

    def __init__(self, root_cause: Exception | None = None):
        """Initialize the exception."""
        super().__init__()
        self.root_cause = root_cause

    def __str__(self) -> str:
        """Return a string representation of the exception."""
        if self.root_cause:
            return f"dbt parse failed.\n\n{self.root_cause!s}"

        return (
            "dbt parse failed. Root cause not found. Please run `dbt parse` manually."
        )

__init__(root_cause=None)

Initialize the exception.

Source code in src/dbt_score/dbt_utils.py

def __init__(self, root_cause: Exception | None = None):
    """Initialize the exception."""
    super().__init__()
    self.root_cause = root_cause

__str__()

Return a string representation of the exception.

Source code in src/dbt_score/dbt_utils.py

def __str__(self) -> str:
    """Return a string representation of the exception."""
    if self.root_cause:
        return f"dbt parse failed.\n\n{self.root_cause!s}"

    return (
        "dbt parse failed. Root cause not found. Please run `dbt parse` manually."
    )

dbt_ls(select)

Run dbt ls.

Source code in src/dbt_score/dbt_utils.py

@dbt_required
def dbt_ls(select: Iterable[str] | None) -> Iterable[str]:
    """Run dbt ls."""
    cmd = [
        "ls",
        "--resource-types",
        "model",
        "source",
        "snapshot",
        "exposure",
        "seed",
        "--output",
        "name",
    ]
    if select:
        cmd += ["--select", *select]

    with _disable_dbt_stdout():
        result: "dbtRunnerResult" = dbtRunner().invoke(cmd)

    if not result.success:
        raise DbtLsException("dbt ls failed.") from result.exception

    selected = cast(Iterable[str], result.result)  # mypy hint
    return selected

dbt_parse()

Parse a dbt project.

Returns:

Type	Description
dbtRunnerResult	The dbt parse run result.

Raises:

Type	Description
DbtParseException	dbt parse failed.

Source code in src/dbt_score/dbt_utils.py

@dbt_required
def dbt_parse() -> "dbtRunnerResult":
    """Parse a dbt project.

    Returns:
        The dbt parse run result.

    Raises:
        DbtParseException: dbt parse failed.
    """
    with _disable_dbt_stdout():
        result: "dbtRunnerResult" = dbtRunner().invoke(["parse"])

    if not result.success:
        raise DbtParseException(root_cause=result.exception)

    return result

dbt_required(func)

Decorator for methods that require dbt to be installed.

Source code in src/dbt_score/dbt_utils.py

def dbt_required(func: Callable[..., Any]) -> Callable[..., Any]:
    """Decorator for methods that require dbt to be installed."""

    @wraps(func)
    def wrapper(*args: Any, **kwargs: Any) -> Any:
        if not DBT_INSTALLED:
            raise DbtNotInstalledException(
                "This option requires dbt to be installed in the same Python"
                "environment as dbt-score."
            )
        return func(*args, **kwargs)

    return wrapper

get_default_manifest_path()

Get the manifest path.

Source code in src/dbt_score/dbt_utils.py

def get_default_manifest_path() -> Path:
    """Get the manifest path."""
    return (
        Path().cwd()
        / os.getenv("DBT_PROJECT_DIR", "")
        / os.getenv("DBT_TARGET_DIR", "target")
        / "manifest.json"
    )