Skip to content

dbt utils

dbt utilities.

DbtLsException

Bases: Exception

Raised when dbt ls fails.

Source code in src/dbt_score/dbt_utils.py 
25
26
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 
17
18
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 
21
22
class DbtParseException(Exception):
    """Raised when dbt parse fails."""

dbt_ls(select)

Run dbt ls.

Source code in src/dbt_score/dbt_utils.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
@dbt_required
def dbt_ls(select: Iterable[str] | None) -> Iterable[str]:
    """Run dbt ls."""
    cmd = ["ls", "--resource-types", "model", "source", "--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 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
@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("dbt parse failed.") from 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 
29
30
31
32
33
34
35
36
37
38
39
40
41
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 
86
87
88
89
90
91
92
93
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"
    )