Add dbt-cloud integration command to dp cli

# CR

Author: Radosław Dziadosz

data_pipelines_cli/cli_commands/dbtcloud.py

@@ -1,12 +1,14 @@
 import json
+import sys
 from typing import Any, Dict
 
 import click
 
 from data_pipelines_cli.dbt_cloud_api_client import DbtCloudApiClient
 from ..cli_constants import BUILD_DIR
 from ..cli_utils import echo_info
-from ..config_generation import read_dictionary_from_config_directory
+from ..config_generation import read_dictionary_from_config_directory, generate_profiles_yml
+from ..dbt_utils import _dump_dbt_vars_from_configs_to_string, run_dbt_command
 
 
 def read_dbtcloud_config() -> Dict[str, Any]:
@@ -33,6 +35,26 @@ def read_bigquery_config(env: str) -> Dict[str, Any]:
     return read_dictionary_from_config_directory(BUILD_DIR.joinpath("dag"), env, "bigquery.yml")
 
 
+def resolve_env_var(unresolved_text, env):
+    """
+    Resolves environment variables and jinja in the given text using the dbt show command.
+    :param unresolved_text: Text to be resolved
+    :param env: Environment dir name
+    :return: Parsed text
+    """
+    profiles_path = generate_profiles_yml(env, False)
+    dbt_command_result_bytes = run_dbt_command(
+        ("show", "--inline", f"SELECT '{unresolved_text}' AS parsed", "--output", "json"),
+        env,
+        profiles_path,
+        log_format_json=True,
+        capture_output=True)
+    decoded_output = dbt_command_result_bytes.stdout.decode(encoding=sys.stdout.encoding or "utf-8")
+    for line in map(json.loads, decoded_output.splitlines()):
+        if line.get('data', {}).get('node_name') == "inline_query":
+            return json.loads(line["data"]["preview"])[0]["parsed"]
+
+
 @click.command(name="configure-cloud", help="Create dbt Cloud project")
 @click.option(
     "--account_id",
@@ -72,29 +94,33 @@ def configure_cloud_command(
     client = DbtCloudApiClient(f"https://cloud.getdbt.com/api", account_id, token)
 
     dbtcloud_config = read_dbtcloud_config()
+    base_bq_config = read_bigquery_config("base")
     file = open(keyfile)
     keyfile_data = json.load(file)
-    project_id = client.create_project(dbtcloud_config["project_name"])
-    (repository_id, deploy_key) = client.create_repository(project_id, remote_url)
+    dbtcloud_project_id = client.create_project(dbtcloud_config["project_name"])
+    (repository_id, deploy_key) = client.create_repository(dbtcloud_project_id, remote_url)
     echo_info("You need to add the generated key text as a deploy key to the target repository.\n"
               "This gives dbt Cloud permissions to read / write in the repository\n"
               f"{deploy_key}")
 
     environments_projects = {}
     for environment in dbtcloud_config["environments"]:
-        environment_id = create_environment(client, environment, project_id)
+        bq_config = read_bigquery_config(environment["config_dir"])
+        environments_projects[environment["name"]] = resolve_env_var(bq_config["project"],
+                                                                     environment["config_dir"])
+        environment_id = create_environment(client, environment, bq_config["dataset"],
+                                            dbtcloud_project_id)
         if environment["type"] == "deployment":
-            client.create_job(project_id, environment_id, dbtcloud_config["schedule_interval"],
-                              "Job - " + environment["name"])
-        bq_config = read_bigquery_config(environment["bq_config_dir"])
-        environments_projects[environment["name"]] = bq_config["project"]
+            dbt_vars = _dump_dbt_vars_from_configs_to_string(environment["config_dir"]).strip()
+            client.create_job(dbtcloud_project_id, environment_id, environment["schedule_interval"],
+                              "Job - " + environment["name"], dbt_vars)
 
-    client.create_environment_variable(project_id, dbtcloud_config["default_gcp_project"],
+    client.create_environment_variable(dbtcloud_project_id, base_bq_config["project"],
                                        environments_projects)
 
-    connection_id = create_bq_connection(client, keyfile_data, project_id)
+    connection_id = create_bq_connection(client, keyfile_data, dbtcloud_project_id)
 
-    client.associate_connection_repository(dbtcloud_config["project_name"], project_id,
+    client.associate_connection_repository(dbtcloud_config["project_name"], dbtcloud_project_id,
                                            connection_id, repository_id)
 
 
@@ -111,7 +137,7 @@ def create_bq_connection(client, keyfile_data, project_id):
         project_id=project_id,
         name="BQ Connection Name",
         is_active=True,
-        gcp_project_id="{{ env_var(\"DBT_GCP_PROJECT\") }} ",
+        gcp_project_id="{{ env_var(\"DBT_GCP_PROJECT\") }}",
         timeout_seconds=100,
         private_key_id=keyfile_data["private_key_id"],
         private_key=keyfile_data["private_key"],
@@ -124,17 +150,18 @@ def create_bq_connection(client, keyfile_data, project_id):
     )
 
 
-def create_environment(client, environment, project_id):
+def create_environment(client, environment, dataset, project_id):
     """
     Creates a dbt Cloud environment with the specified configuration
 
     :param client: API Client
     :param environment: Config of environment to be created
     :param project_id: ID of the project
+    :param dataset: Name of target dataset
     :return: ID of created environment
     """
     if environment["type"] == "deployment":
-        credentials_id = client.create_credentials(environment["dataset"], project_id)
+        credentials_id = client.create_credentials(dataset, project_id)
     else:
         credentials_id = None
     environment_id = client.create_environment(project_id, environment["type"], environment["name"],

data_pipelines_cli/cli_commands/publish.py

@@ -5,9 +5,9 @@
 
 import click
 import yaml
-from dbt.contracts.graph.compiled import ManifestNode
 from dbt.contracts.graph.manifest import Manifest
-from dbt.contracts.graph.parsed import ColumnInfo
+from dbt.contracts.graph.nodes import ColumnInfo
+from dbt.contracts.graph.nodes import ManifestNode
 
 from ..cli_constants import BUILD_DIR
 from ..cli_utils import echo_info, echo_warning

data_pipelines_cli/dbt_cloud_api_client.py

@@ -137,7 +137,6 @@ def create_environment_variable(self, project_id, default, environments):
         new_env = {
             "env_var": env_var
         }
-        print(new_env)
         new_env_data = json.dumps(new_env)
 
         response = self.request(
@@ -260,16 +259,18 @@ def create_bigquery_connection(
 
         return response["data"]["id"]
 
-    def create_job(self, project_id, environment_id, schedule_cron, name):
+    def create_job(self, project_id, environment_id, schedule_cron, name, vars):
         """
         Creates sample job for given project and environment. Job is triggered by the scheduler
         executes commands: dbt seed, dbt test and dbt run.
         :param project_id: ID of the project
         :param environment_id: ID of the environment
         :param schedule_cron: Schedule (cron syntax)
         :param name: Name of the job
+        :param vars: Variables passed to commands
         :return: ID of created job
         """
+
         job_details = {
             "account_id": self.account_id,
             "project_id": project_id,
@@ -282,9 +283,9 @@ def create_job(self, project_id, environment_id, schedule_cron, name):
                 "github_webhook": False
             },
             "execute_steps": [
-                "dbt seed",
-                "dbt test",
-                "dbt run"
+                "dbt seed --vars '" + vars + "'",
+                "dbt run --vars '" + vars + "'",
+                "dbt test --vars '" + vars + "'"
             ],
             "settings": {
                 "threads": 1,

docs/configuration.rst

@@ -392,12 +392,6 @@ dbt Cloud configuration
    * - project_name
      - string
      - Name of the project to be created in dbt Cloud
-   * - schedule_interval
-     - string
-     - The cron expression with which the example job will be run
-   * - default_gcp_project
-     - string
-     - GCP project that will be used by default if a new environment is created in dbt Cloud
    * - environments
      - Array
      - Details of the environments to be created in dbt Cloud
@@ -414,31 +408,31 @@ Configuration of the environments:
    * - name
      - string
      - Name of the environment that will be created in dbt Cloud
-   * - dataset
+   * - type
      - string
-     - Target dataset for this environment
+     - In dbt Cloud, there are two types of environments: deployment and development. Deployment environments determine the settings used when jobs created within that environment are executed. Development environments determine the settings used in the dbt Cloud IDE for that particular dbt Cloud Project. Each dbt Cloud project can only have a single development environment but can have any number of deployment environments.
    * - dbt_version
      - string
      - The dbt version used in this environment
-   * - bq_config_dir
+   * - schedule_interval
+     - string
+     - The cron expression with which the example job will be run. This setting is only needed for the deployment environment.
+   * - config_dir
      - string
-     - The name of the dp env directory where the bigquery configuration for the environment is located. The name of the project in GCP will be read from it.
+     - The name of the dp env directory where the bigquery configuration for the environment is located. The name of the project in GCP and target dataset will be read from it.
 
 Example ``dbtcloud.yml`` file might look like this:
 
 .. code-block:: yaml
 
     project_name: "Data Pipelines Project"
-    schedule_interval: "0 12 * * *"
-    default_gcp_project: "default-project"
     environments:
       - name: "Develop"
-        dataset: "dev"
         dbt_version: "1.0.0"
         type: "development"
-        bq_config_dir: "dev"
+        config_dir: "dev"
       - name: "Production"
-        dataset: "prod"
         dbt_version: "1.0.0"
         type: "deployment"
-        bq_config_dir: "prod"
+        config_dir: "prod"
+        schedule_interval: "0 12 * * *"

setup.py

@@ -16,7 +16,7 @@
     "fsspec==2022.11.0",
     "packaging==21.3",
     "colorama==0.4.5",
-    "dbt-core==1.3.1",
+    "dbt-core==1.5.0",
 ]
 
 EXTRA_FILESYSTEMS_REQUIRE = {