feat(dbt-cloud): disable schedule only if it exists (#11690)

### Summary & Motivation We don't need to always disable the schedule. Disable it only if it exists. This optimization is required in the future when we check the dbt Cloud job's update time to determine whether or not to invalidate its reference to run artifacts from a compile run. ### How I Tested These Changes local, bk
dagster-io · Jan 17, 2023 · 41d6c4d · 41d6c4d
1 parent 789ac1a
commit 41d6c4d
Show file tree

Hide file tree

Showing 3 changed files with 31 additions and 4 deletions.
diff --git a/python_modules/libraries/dagster-dbt/dagster_dbt/cloud/resources.py b/python_modules/libraries/dagster-dbt/dagster_dbt/cloud/resources.py
@@ -229,13 +229,16 @@ def run_job(self, job_id: int, **kwargs) -> Mapping[str, Any]:
         Returns:
             Dict[str, Any]: Parsed json data from the response to this request
         """
-        if self._disable_schedule_on_trigger:
-            self._log.info("Disabling dbt Cloud job schedule.")
-            self.update_job(job_id, triggers={"schedule": False})
         self._log.info(f"Initializing run for job with job_id={job_id}")
         if "cause" not in kwargs:
             kwargs["cause"] = "Triggered via Dagster"
         resp = self.make_request("POST", f"{self._account_id}/jobs/{job_id}/run/", data=kwargs)
+
+        has_schedule: bool = resp.get("job", {}).get("triggers", {}).get("schedule", False)
+        if has_schedule and self._disable_schedule_on_trigger:
+            self._log.info("Disabling dbt Cloud job schedule.")
+            self.update_job(job_id, triggers={"schedule": False})
+
         self._log.info(
             f"Run initialized with run_id={resp['id']}. View this run in "
             f"the dbt Cloud UI: {resp['href']}"

diff --git a/python_modules/libraries/dagster-dbt/dagster_dbt_tests/cloud/test_resources.py b/python_modules/libraries/dagster-dbt/dagster_dbt_tests/cloud/test_resources.py
@@ -170,6 +170,30 @@ def test_no_disable_schedule():
         # endpoint for disabling the schedule has not been set up, so will fail if attempted
         dc_resource.run_job_and_poll(SAMPLE_JOB_ID)
 
+    # If the schedule was already disabled, no need to re-disable it.
+    dc_resource = get_dbt_cloud_resource()
+    with responses.RequestsMock() as rsps:
+        # endpoint for launching run
+        rsps.add(
+            rsps.POST,
+            f"{SAMPLE_API_PREFIX}/jobs/{SAMPLE_JOB_ID}/run/",
+            json=sample_run_details(job={"triggers": {"schedule": False}}),
+        )
+        # run will immediately succeed
+        rsps.add(
+            rsps.GET,
+            f"{SAMPLE_API_PREFIX}/runs/{SAMPLE_RUN_ID}/",
+            json=sample_run_details(status_humanized="Success"),
+        )
+        # endpoint for run_results.json
+        rsps.add(
+            rsps.GET,
+            f"{SAMPLE_API_PREFIX}/runs/{SAMPLE_RUN_ID}/artifacts/run_results.json",
+            json=sample_run_results(),
+        )
+        # endpoint for disabling the schedule has not been set up, so will fail if attempted
+        dc_resource.run_job_and_poll(SAMPLE_JOB_ID)
+
 
 @pytest.mark.parametrize(
     "final_status,expected_behavior",

diff --git a/python_modules/libraries/dagster-dbt/dagster_dbt_tests/cloud/utils.py b/python_modules/libraries/dagster-dbt/dagster_dbt_tests/cloud/utils.py
@@ -138,7 +138,7 @@ def sample_run_details(include_related=None, **kwargs):
         "last_heartbeat_at": None,
         "should_start_at": "2021-11-01 22:47:48.501943+00:00",
         "trigger": None,
-        "job": None,
+        "job": {"triggers": {"schedule": True}},
         "environment": None,
         "run_steps": [],
         "status_humanized": "Success",