Skip to content

dbt utils

dbt utilities.

DbtLsException

Bases: Exception

Raised when dbt ls fails.

Source code in src/dbt_score/dbt_utils.py 
15
16
class DbtLsException(Exception):
    """Raised when dbt ls fails."""

DbtParseException

Bases: Exception

Raised when dbt parse fails.

Source code in src/dbt_score/dbt_utils.py 
11
12
class DbtParseException(Exception):
    """Raised when dbt parse fails."""

dbt_ls(select)

Run dbt ls.

Source code in src/dbt_score/dbt_utils.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def dbt_ls(select: Iterable[str] | None) -> Iterable[str]:
    """Run dbt ls."""
    cmd = ["ls", "--resource-type", "model", "--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 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
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

get_default_manifest_path()

Get the manifest path.

Source code in src/dbt_score/dbt_utils.py 
59
60
61
62
63
64
65
66
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"
    )