Feature: implement dbt deps and dbt vars

examples/sample_project/profiles.yml

@@ -13,3 +13,6 @@ sample_project:
     override_in_test:
       type: duckdb
       path: /does/not/exist/so/override.duckdb
+    vars_test:
+      type: "{{ var('adapter_type') }}"
+      path: "{{ var('duckdb_db_path') }}"

prefect_dbt_flow/dbt/__init__.py

@@ -74,8 +74,12 @@ class DbtDagOptions:
         select: dbt module to include in the run
         exclude: dbt module to exclude in the run
         run_test_after_model: run test afeter run model
+        vars: dbt vars
+        install_deps: install dbt dependencies, default behavior install deps
     """
 
     select: Optional[str] = None
     exclude: Optional[str] = None
     run_test_after_model: bool = False
+    vars: Optional[dict[str, str]] = None
+    install_deps: bool = True

prefect_dbt_flow/dbt/cli.py

@@ -1,4 +1,5 @@
 """Utility functions for interacting with dbt using command-line commands."""
+import json
 import shutil
 from typing import Optional
 
@@ -11,6 +12,7 @@
 def dbt_ls(
     project: DbtProject,
     dag_options: Optional[DbtDagOptions],
+    profile: Optional[DbtProfile],
     output: str = "json",
 ) -> str:
     """
@@ -19,6 +21,7 @@ def dbt_ls(
     Args:
         project: A class that represents a dbt project configuration.
         dag_options: A class to add dbt DAG configurations.
+        profile: A class that represents a dbt profile configuration.
         output: Format of output, default is JSON.
 
     Returns:
@@ -29,11 +32,16 @@ def dbt_ls(
     dbt_ls_cmd.extend(["--profiles-dir", str(project.profiles_dir)])
     dbt_ls_cmd.extend(["--output", output])
 
+    if profile:
+        dbt_ls_cmd.extend(["-t", profile.target])
+
     if dag_options:
         if dag_options.select:
             dbt_ls_cmd.extend(["--select", dag_options.select])
         if dag_options.exclude:
             dbt_ls_cmd.extend(["--exclude", dag_options.exclude])
+        if dag_options.vars:
+            dbt_ls_cmd.extend(["--vars", f"'{json.dumps(dag_options.vars)}'"])
 
     return cmd.run(" ".join(dbt_ls_cmd))
 
@@ -42,6 +50,7 @@ def dbt_run(
     project: DbtProject,
     model: str,
     profile: Optional[DbtProfile],
+    dag_options: Optional[DbtDagOptions],
 ) -> str:
     """
     Function that executes `dbt run` command
@@ -50,6 +59,7 @@ def dbt_run(
         project: A class that represents a dbt project configuration.
         model: Name of the model to run.
         profile: A class that represents a dbt profile configuration.
+        dag_options: A class to add dbt DAG configurations.
 
     Returns:
         A string representing the output of the `dbt run` command.
@@ -62,13 +72,18 @@ def dbt_run(
     if profile:
         dbt_run_cmd.extend(["-t", profile.target])
 
+    if dag_options:
+        if dag_options.vars:
+            dbt_run_cmd.extend(["--vars", f"'{json.dumps(dag_options.vars)}'"])
+
     return cmd.run(" ".join(dbt_run_cmd))
 
 
 def dbt_test(
     project: DbtProject,
     model: str,
     profile: Optional[DbtProfile],
+    dag_options: Optional[DbtDagOptions],
 ) -> str:
     """
     Function that executes `dbt test` command
@@ -77,6 +92,7 @@ def dbt_test(
         project: A class that represents a dbt project configuration.
         model: Name of the model to run.
         profile: A class that represents a dbt profile configuration.
+        dag_options: A class to add dbt DAG configurations.
 
     Returns:
         A string representing the output of the `dbt test` command.
@@ -89,13 +105,18 @@ def dbt_test(
     if profile:
         dbt_test_cmd.extend(["-t", profile.target])
 
+    if dag_options:
+        if dag_options.vars:
+            dbt_test_cmd.extend(["--vars", f"'{json.dumps(dag_options.vars)}'"])
+
     return cmd.run(" ".join(dbt_test_cmd))
 
 
 def dbt_seed(
     project: DbtProject,
     seed: str,
     profile: Optional[DbtProfile],
+    dag_options: Optional[DbtDagOptions],
 ) -> str:
     """
     Function that executes `dbt seed` command
@@ -104,7 +125,7 @@ def dbt_seed(
         project: A class that represents a dbt project configuration.
         seed: Name of the seed to run.
         profile: A class that represents a dbt profile configuration.
-
+        dag_options: A class to add dbt DAG configurations.
 
     Returns:
         A string representing the output of the `dbt seed` command
@@ -117,13 +138,18 @@ def dbt_seed(
     if profile:
         dbt_seed_cmd.extend(["-t", profile.target])
 
+    if dag_options:
+        if dag_options.vars:
+            dbt_seed_cmd.extend(["--vars", f"'{json.dumps(dag_options.vars)}'"])
+
     return cmd.run(" ".join(dbt_seed_cmd))
 
 
 def dbt_snapshot(
     project: DbtProject,
     snapshot: str,
     profile: Optional[DbtProfile],
+    dag_options: Optional[DbtDagOptions],
 ) -> str:
     """
     Function that executes `dbt snapshot` command
@@ -132,17 +158,51 @@ def dbt_snapshot(
         project: A class that represents a dbt project configuration.
         snapshot: Name of the snapshot to run.
         profile: A class that represents a dbt profile configuration.
-
+        dag_options: A class to add dbt DAG configurations.
 
     Returns:
         A string representing the output of the `dbt snapshot` command
     """
-    dbt_seed_cmd = [DBT_EXE, "snapshot"]
-    dbt_seed_cmd.extend(["--project-dir", str(project.project_dir)])
-    dbt_seed_cmd.extend(["--profiles-dir", str(project.profiles_dir)])
-    dbt_seed_cmd.extend(["--select", snapshot])
+    dbt_snapshot_cmd = [DBT_EXE, "snapshot"]
+    dbt_snapshot_cmd.extend(["--project-dir", str(project.project_dir)])
+    dbt_snapshot_cmd.extend(["--profiles-dir", str(project.profiles_dir)])
+    dbt_snapshot_cmd.extend(["--select", snapshot])
 
     if profile:
-        dbt_seed_cmd.extend(["-t", profile.target])
+        dbt_snapshot_cmd.extend(["-t", profile.target])
 
-    return cmd.run(" ".join(dbt_seed_cmd))
+    if dag_options:
+        if dag_options.vars:
+            dbt_snapshot_cmd.extend(["--vars", f"'{json.dumps(dag_options.vars)}'"])
+
+    return cmd.run(" ".join(dbt_snapshot_cmd))
+
+
+def dbt_deps(
+    project: DbtProject,
+    profile: Optional[DbtProfile],
+    dag_options: Optional[DbtDagOptions],
+) -> str:
+    """
+    Function that executes `dbt deps` command
+
+    Args:
+        project: A class that represents a dbt project configuration.
+        profile: A class that represents a dbt profile configuration.
+        dag_options: A class to add dbt DAG configurations.
+
+    Returns:
+        A string representing the output of the `dbt deps` command
+    """
+    dbt_deps_cmd = [DBT_EXE, "deps"]
+    dbt_deps_cmd.extend(["--project-dir", str(project.project_dir)])
+    dbt_deps_cmd.extend(["--profiles-dir", str(project.profiles_dir)])
+
+    if profile:
+        dbt_deps_cmd.extend(["-t", profile.target])
+
+    if dag_options:
+        if dag_options.vars:
+            dbt_deps_cmd.extend(["--vars", f"'{json.dumps(dag_options.vars)}'"])
+
+    return cmd.run(" ".join(dbt_deps_cmd))

prefect_dbt_flow/dbt/graph.py

@@ -5,20 +5,25 @@
 from prefect_dbt_flow.dbt import (
     DbtDagOptions,
     DbtNode,
+    DbtProfile,
     DbtProject,
     DbtResourceType,
     cli,
 )
+from prefect_dbt_flow.dbt.profile import override_profile
 
 
 def parse_dbt_project(
-    project: DbtProject, dag_options: Optional[DbtDagOptions] = None
+    project: DbtProject,
+    profile: Optional[DbtProfile],
+    dag_options: Optional[DbtDagOptions] = None,
 ) -> List[DbtNode]:
     """
     Parses a list of dbt nodes class objects from dbt ls cli command.
 
     Args:
         project: A class that represents a dbt project configuration.
+        profile: A class that represents a dbt profile configuration.
         dag_options: A class to add dbt DAG configurations.
 
     Returns:
@@ -27,7 +32,11 @@ def parse_dbt_project(
     dbt_graph: List[DbtNode] = []
     models_with_tests: List[str] = []
 
-    dbt_ls_output = cli.dbt_ls(project, dag_options)
+    with override_profile(project, profile) as _project:
+        if not dag_options or dag_options.install_deps:
+            cli.dbt_deps(_project, profile, dag_options)
+
+        dbt_ls_output = cli.dbt_ls(_project, dag_options, profile)
 
     for line in dbt_ls_output.split("\n"):
         try: