Merge pull request #135 from dbt-labs/add-json-output

Add `--json` parameter to `plan` and `sync`

Author: b-per

docs/advanced_config/jobs_importing.md

@@ -73,7 +73,7 @@ dbt-jobs-as-code import-jobs --account-id 1234 --project-id 3213 --environment-i
 It would then be possible to automate the creation of PRs whenever the `jobs.yml` file is updated, meaning that some jobs would have been updated in the dbt Cloud UI. The GitHub action [Create Pull Request](https://github.com/marketplace/actions/create-pull-request) could be used to implement this flow.
 
 
-Then, the `jobs.yml` file can be used to import the jobs in a different environment with the following command, like described in [YAML templating](templating.md) and in the [typical flows](../typical_flows.md) page :
+Then, the `jobs.yml` file can be used to import the jobs in a different environment with the following command, like described in [YAML templating](templating.md) and in the [typical flows](../typical_flows.md#advanced-flows) page :
 
 ```bash
 dbt-jobs-as-code plan jobs.yml -v qa_vars.yml --limit-projects-envs-to-yml

docs/changelog.md

@@ -1,6 +1,10 @@
 
 To see the details of all changes, head to the GitHub repo
 
+### 1.5
+
+- Add `--json` to `plan` and `sync` to output the `stdout` changes in JSON format. This can be useful for automating some processes and consuming the changes from scripts. We are still printing logs to `stderr` though, so to remove those logs you can redirect `stderr` to `/dev/null` or redirect `stdout` to a file and then read from the file.
+
 ### 1.4
 
 - Add `--templated-fields` to `import-jobs` to add Jinja variables to the generated YAML file. This can be useful to allow users to maintain jobs in the dbt Cloud UI and set a process to automatically promote those to other environments.

src/dbt_jobs_as_code/cloud_yaml_mapping/change_set.py

@@ -1,12 +1,15 @@
 import glob
+import json
 import os
 import string
 from collections import Counter
+from typing import Dict, Optional
 
 from beartype import BeartypeConf, BeartypeStrategy, beartype
 from beartype.typing import Callable, List
 from loguru import logger
 from pydantic import BaseModel
+from rich.console import Console
 from rich.table import Table
 
 from dbt_jobs_as_code.client import DBTCloud, DBTCloudException
@@ -18,6 +21,12 @@
 nobeartype = beartype(conf=BeartypeConf(strategy=BeartypeStrategy.O0))
 
 
+def json_serializer_type(obj):
+    if isinstance(obj, type):
+        return obj.__name__
+    raise TypeError(f"Object of type {type(obj).__name__} is not JSON serializable")
+
+
 class Change(BaseModel):
     """Describes what a given change is and how to apply it."""
 
@@ -28,6 +37,7 @@ class Change(BaseModel):
     env_id: int
     sync_function: Callable
     parameters: dict
+    differences: Optional[Dict] = {}
 
     def __str__(self):
         return f"{self.action.upper()} {string.capwords(self.type)} {self.identifier}"
@@ -74,6 +84,32 @@ def to_table(self) -> Table:
 
         return table
 
+    def to_json(self) -> dict:
+        """Return a structured JSON representation of the changeset."""
+        job_changes = []
+        env_var_changes = []
+
+        for change in self.root:
+            # Create the base change dictionary for overall changes
+            overall_change_dict = {
+                "action": change.action.upper(),
+                "type": string.capwords(change.type),
+                "identifier": change.identifier,
+                "project_id": change.proj_id,
+                "environment_id": change.env_id,
+                "differences": change.differences,
+            }
+
+            if change.type == "job":
+                job_changes.append(overall_change_dict)
+            elif change.type == "env var overwrite":
+                env_var_changes.append(overall_change_dict)
+
+        return {
+            "job_changes": job_changes,
+            "env_var_overwrite_changes": env_var_changes,
+        }
+
     def __len__(self):
         return len(self.root)
 
@@ -144,6 +180,7 @@ def build_change_set(
     project_ids: List[int],
     environment_ids: List[int],
     limit_projects_envs_to_yml: bool = False,
+    output_json: bool = False,
 ):
     """Compares the config of YML files versus dbt Cloud.
     Depending on the value of no_update, it will either update the dbt Cloud config or not.
@@ -203,12 +240,15 @@ def build_change_set(
     deleted_jobs = set(tracked_jobs.keys()) - set(defined_jobs.keys())
 
     # Update changed jobs
-    logger.info("Detected {count} existing jobs.", count=len(shared_jobs))
+    if not output_json:
+        logger.info("Detected {count} existing jobs.", count=len(shared_jobs))
     for identifier in shared_jobs:
-        logger.info("Checking for differences in {identifier}", identifier=identifier)
-        if not check_job_mapping_same(
+        if not output_json:
+            logger.info("Checking for differences in {identifier}", identifier=identifier)
+        is_same, diff_data = check_job_mapping_same(
             source_job=defined_jobs[identifier], dest_job=tracked_jobs[identifier]
-        ):
+        )
+        if not is_same:
             dbt_cloud_change = Change(
                 identifier=identifier,
                 type="job",
@@ -217,12 +257,21 @@ def build_change_set(
                 env_id=defined_jobs[identifier].environment_id,
                 sync_function=dbt_cloud.update_job,
                 parameters={"job": defined_jobs[identifier]},
+                differences=diff_data.get("differences", {}) if diff_data else {},
             )
             dbt_cloud_change_set.append(dbt_cloud_change)
             defined_jobs[identifier].id = tracked_jobs[identifier].id
+            if not output_json:
+                console = Console()
+                console.print(
+                    f"❌ Job {identifier} is different - Diff:\n{json.dumps(diff_data, indent=2, default=json_serializer_type)}"
+                )
+        elif not output_json:
+            logger.success(f"✅ Job {identifier} is identical")
 
     # Create new jobs
-    logger.info("Detected {count} new jobs.", count=len(created_jobs))
+    if not output_json:
+        logger.info("Detected {count} new jobs.", count=len(created_jobs))
     for identifier in created_jobs:
         dbt_cloud_change = Change(
             identifier=identifier,
@@ -236,7 +285,8 @@ def build_change_set(
         dbt_cloud_change_set.append(dbt_cloud_change)
 
     # Remove Deleted Jobs
-    logger.info("Detected {count} deleted jobs.", count=len(deleted_jobs))
+    if not output_json:
+        logger.info("Detected {count} deleted jobs.", count=len(deleted_jobs))
     for identifier in deleted_jobs:
         dbt_cloud_change = Change(
             identifier=identifier,
@@ -252,7 +302,8 @@ def build_change_set(
     # -- ENV VARS --
     # Now that we have replicated all jobs we can get their IDs for further API calls
     mapping_job_identifier_job_id = dbt_cloud.build_mapping_job_identifier_job_id(cloud_jobs)
-    logger.debug(f"Mapping of job identifier to id: {mapping_job_identifier_job_id}")
+    if not output_json:
+        logger.debug(f"Mapping of job identifier to id: {mapping_job_identifier_job_id}")
 
     # Replicate the env vars from the YML to dbt Cloud
     for job in defined_jobs.values():
@@ -261,14 +312,21 @@ def build_change_set(
             all_env_vars_for_job = dbt_cloud.get_env_vars(project_id=job.project_id, job_id=job_id)
             for env_var_yml in job.custom_environment_variables:
                 env_var_yml.job_definition_id = job_id
-                same_env_var, env_var_id = check_env_var_same(
+                same_env_var, env_var_id, diff_data = check_env_var_same(
                     source_env_var=env_var_yml, dest_env_vars=all_env_vars_for_job
                 )
-                if not same_env_var:
+                if not same_env_var and diff_data:
+                    action = (
+                        "CREATE"
+                        if diff_data.get("old_value") is None
+                        else "DELETE"
+                        if diff_data.get("new_value") is None
+                        else "UPDATE"
+                    )
                     dbt_cloud_change = Change(
                         identifier=f"{job.identifier}:{env_var_yml.name}",
                         type="env var overwrite",
-                        action="update",
+                        action=action,
                         proj_id=job.project_id,
                         env_id=job.environment_id,
                         sync_function=dbt_cloud.update_env_var,
@@ -278,6 +336,7 @@ def build_change_set(
                             "custom_env_var": env_var_yml,
                             "env_var_id": env_var_id,
                         },
+                        differences=diff_data,
                     )
                     dbt_cloud_change_set.append(dbt_cloud_change)
 
@@ -315,7 +374,8 @@ def build_change_set(
             for env_var, env_var_val in env_var_dbt_cloud.items():
                 # If the env var is not in the YML but is defined at the "job" level in dbt Cloud, we delete it
                 if env_var not in env_vars_for_job and env_var_val.id:
-                    logger.info(f"{env_var} not in the YML file but in the dbt Cloud job")
+                    if not output_json:
+                        logger.info(f"{env_var} not in the YML file but in the dbt Cloud job")
                     dbt_cloud_change = Change(
                         identifier=f"{job.identifier}:{env_var}",
                         type="env var overwrite",
@@ -332,6 +392,5 @@ def build_change_set(
 
     # Filtering out the change set, if project_id(s), environment_id(s) are passed as arguments to function
     # TODO: Confirm if this is the desired functionality, remove otherwise
-    logger.debug(f"dbt cloud change set: {dbt_cloud_change_set}")
 
     return dbt_cloud_change_set

src/dbt_jobs_as_code/loader/load.py

@@ -73,6 +73,14 @@ def _load_yaml_no_template(config_files: List[str]) -> dict:
     return combined_config
 
 
+def _replace_none_with_null(obj):
+    if isinstance(obj, dict):
+        return {k: _replace_none_with_null(v) for k, v in obj.items()}
+    elif isinstance(obj, list):
+        return [_replace_none_with_null(item) for item in obj]
+    return "null" if obj is None else obj
+
+
 def _load_vars_files(vars_file: List[str]) -> dict:
     """Load and merge multiple vars files into a single dictionary.
 
@@ -97,7 +105,7 @@ def _load_vars_files(vars_file: List[str]) -> dict:
                         f"Variable '{key}' is defined multiple times in vars files"
                     )
             template_vars_values.update(vars_data)
-    return template_vars_values
+    return _replace_none_with_null(template_vars_values)  # type: ignore
 
 
 def _load_yaml_with_template(config_files: List[str], vars_file: List[str]) -> dict:

src/dbt_jobs_as_code/main.py

@@ -1,3 +1,4 @@
+import json
 import os
 import sys
 from importlib.metadata import version
@@ -10,7 +11,7 @@
 from ruamel.yaml import YAML
 
 from dbt_jobs_as_code.client import DBTCloud
-from dbt_jobs_as_code.cloud_yaml_mapping.change_set import build_change_set
+from dbt_jobs_as_code.cloud_yaml_mapping.change_set import build_change_set, json_serializer_type
 from dbt_jobs_as_code.cloud_yaml_mapping.validate_link import can_be_linked
 from dbt_jobs_as_code.exporter.export import export_jobs_yml
 from dbt_jobs_as_code.importer import check_job_fields, fetch_jobs, get_account_id
@@ -58,6 +59,13 @@
     help="The path to your vars_yml YML file (or pattern for those files) when using a templated job YML file.",
 )
 
+option_json_output = click.option(
+    "--json",
+    "output_json",
+    is_flag=True,
+    help="Output results in JSON format instead of human-readable text.",
+)
+
 
 @click.group(
     help=f"dbt-jobs-as-code {VERSION}\n\nA CLI to allow defining dbt Cloud jobs as code",
@@ -75,13 +83,15 @@ def cli() -> None:
 @option_project_ids
 @option_environment_ids
 @option_limit_projects_envs_to_yml
+@option_json_output
 def sync(
     config: str,
     vars_yml,
     project_id,
     environment_id,
     limit_projects_envs_to_yml,
     disable_ssl_verification,
+    output_json: bool,
 ):
     """Synchronize a dbt Cloud job config file against dbt Cloud.
     This command will update dbt Cloud with the changes in the local YML file. It is recommended to run a `plan` first to see what will be changed.
@@ -111,13 +121,20 @@ def sync(
         cloud_project_ids,
         cloud_environment_ids,
         limit_projects_envs_to_yml,
+        output_json=output_json,
     )
     if len(change_set) == 0:
-        logger.success("-- SYNC -- No changes detected.")
+        if output_json:
+            print(json.dumps({"job_changes": [], "env_var_overwrite_changes": []}))
+        else:
+            logger.success("-- SYNC -- No changes detected.")
     else:
-        logger.info("-- SYNC -- {count} changes detected.", count=len(change_set))
-        console = Console()
-        console.log(change_set.to_table())
+        if output_json:
+            print(json.dumps(change_set.to_json()))
+        else:
+            logger.info("-- SYNC -- {count} changes detected.", count=len(change_set))
+            console = Console()
+            console.log(change_set.to_table())
     change_set.apply()
 
     if not change_set.apply_success:
@@ -132,13 +149,15 @@ def sync(
 @option_project_ids
 @option_environment_ids
 @option_limit_projects_envs_to_yml
+@option_json_output
 def plan(
     config: str,
     vars_yml: str,
     project_id: List[int],
     environment_id: List[int],
     limit_projects_envs_to_yml: bool,
     disable_ssl_verification: bool,
+    output_json: bool,
 ):
     """Check the difference between a local file and dbt Cloud without updating dbt Cloud.
     This command will not update dbt Cloud.
@@ -167,13 +186,20 @@ def plan(
         cloud_project_ids,
         cloud_environment_ids,
         limit_projects_envs_to_yml,
+        output_json=output_json,
     )
     if len(change_set) == 0:
-        logger.success("-- PLAN -- No changes detected.")
+        if output_json:
+            print(json.dumps({"job_changes": [], "env_var_overwrite_changes": []}))
+        else:
+            logger.success("-- PLAN -- No changes detected.")
     else:
-        logger.info("-- PLAN -- {count} changes detected.", count=len(change_set))
-        console = Console()
-        console.log(change_set.to_table())
+        if output_json:
+            print(json.dumps(change_set.to_json(), default=json_serializer_type))
+        else:
+            logger.info("-- PLAN -- {count} changes detected.", count=len(change_set))
+            console = Console()
+            console.log(change_set.to_table())
 
 
 @cli.command()