Allow deferring to env id

Author: b-per

example_jobs_file/jobs.yml

@@ -32,7 +32,8 @@ jobs:
   job2:
     account_id: 43791
     dbt_version: null
-    deferring_job_definition_id: 43791
+    deferring_job_definition_id: null
+    deferring_environment_id: 43791
     environment_id: 134459
     execute_steps:
     - dbt run-operation clone_all_production_schemas

example_jobs_file/jobs_with_anchors.yml

@@ -8,14 +8,18 @@ anchors:
       hours: null
       interval: 1
       type: every_hour
+  ids_prod_env: &ids_prod_env
+    account_id: 43791
+    environment_id: 134459
+    project_id: 176941
 
 jobs:
   job1: &main_job # using parameters of a job as the anchor
-    account_id: 43791
+    <<: *ids_prod_env
     dbt_version: null
     deactivated: false
     deferring_job_definition_id: null
-    environment_id: 134459
+    deferring_environment_id: null
     execute_steps:
     - "dbt run --select model1+"
     - "dbt run --select model2+"
@@ -25,7 +29,6 @@ jobs:
     generate_docs: false
     generate_sources: true
     name: "My Job 1 with a new name"
-    project_id: 176941
     run_generate_sources: true
     schedule:
       cron: "0 */2 * * *"
@@ -48,6 +51,7 @@ jobs:
   job2:
     <<: *main_job # << means that we take all the values from the first job but we then overwrite them
     deferring_job_definition_id: null
+    deferring_environment_id: null
     execute_steps:
     - dbt run-operation clone_all_production_schemas
     - dbt compile

src/client/__init__.py

@@ -83,6 +83,7 @@ def create_job(self, job: JobDefinition) -> JobDefinition:
 
         if response.status_code >= 400:
             logger.error(response.json())
+            return None
 
         logger.success("Job created successfully.")
 

src/main.py

@@ -268,6 +268,24 @@ def validate(config, online):
             )
             online_check_issues = True
 
+    # In case deferral jobs are mentioned, check that they exist
+    deferral_envs = set(
+        [
+            job.deferring_environment_id
+            for job in defined_jobs
+            if job.deferring_environment_id
+        ]
+    )
+    if deferral_envs:
+        logger.info(f"Checking that Deferring Env IDs are valid")
+        cloud_envs = dbt_cloud.get_environments()
+        cloud_envs_ids = set([env["id"] for env in cloud_envs])
+        if deferral_envs - cloud_envs_ids:
+            logger.error(
+                f"❌ The following deferral environment IDs are not valid: {deferral_envs - cloud_envs_ids}"
+            )
+            online_check_issues = True
+
     if online_check_issues:
         # return an error to handle with bash/CI
         sys.exit(1)

src/schemas/job.py

@@ -21,6 +21,7 @@ class JobDefinition(pydantic.BaseModel):
     settings: Settings
     execution: Execution = Execution()
     deferring_job_definition_id: Optional[int]
+    deferring_environment_id: Optional[int]
     run_generate_sources: bool
     execute_steps: List[str]
     generate_docs: bool

src/schemas/load_job_schema.json

@@ -127,6 +127,10 @@
           "title": "Deferring Job Definition Id",
           "type": ["integer", "null"]
         },
+        "deferring_environment_id": {
+          "title": "Deferring Environment Id",
+          "type": ["integer", "null"]
+        },
         "run_generate_sources": {
           "title": "Run Generate Sources",
           "type": "boolean"

tests/exporter/test_export.py

@@ -26,6 +26,7 @@ def test_export_jobs_yml(capsys):
     execution:
       timeout_seconds: 0
     deferring_job_definition_id:
+    deferring_environment_id:
     run_generate_sources: true
     execute_steps:
     - dbt source freshness
@@ -52,6 +53,7 @@ def test_export_jobs_yml(capsys):
         settings=Settings(threads=4, target_name="production"),
         execution=Execution(timeout_seconds=0),
         deferring_job_definition_id=None,
+        deferring_environment_id=None,
         run_generate_sources=True,
         execute_steps=[
             "dbt source freshness",

tests/loader/jobs.yml

@@ -30,7 +30,8 @@ jobs:
   job2:
     account_id: 43791
     dbt_version: null
-    deferring_job_definition_id: 43791
+    deferring_job_definition_id: null
+    deferring_environment_id: 43791
     environment_id: 134459
     execute_steps:
     - dbt run-operation clone_all_production_schemas

tests/loader/test_loader.py

@@ -28,6 +28,7 @@ def test_import_yml_no_anchor():
                 "settings": {"threads": 4, "target_name": "production"},
                 "execution": {"timeout_seconds": 0},
                 "deferring_job_definition_id": None,
+                "deferring_environment_id": None,
                 "run_generate_sources": True,
                 "execute_steps": [
                     "dbt run --select model1+",
@@ -58,7 +59,8 @@ def test_import_yml_no_anchor():
                 "name": "CI/CD run",
                 "settings": {"threads": 4, "target_name": "TEST"},
                 "execution": {"timeout_seconds": 0},
-                "deferring_job_definition_id": 43791,
+                "deferring_job_definition_id": None,
+                "deferring_environment_id": 43791,
                 "run_generate_sources": False,
                 "execute_steps": [
                     "dbt run-operation clone_all_production_schemas",
@@ -119,6 +121,7 @@ def test_import_yml_anchors():
                 "settings": {"threads": 4, "target_name": "production"},
                 "execution": {"timeout_seconds": 0},
                 "deferring_job_definition_id": None,
+                "deferring_environment_id": None,
                 "run_generate_sources": True,
                 "execute_steps": [
                     "dbt run --select model1+",
@@ -151,6 +154,7 @@ def test_import_yml_anchors():
                 "settings": {"threads": 4, "target_name": "TEST"},
                 "execution": {"timeout_seconds": 0},
                 "deferring_job_definition_id": None,
+                "deferring_environment_id": None,
                 "run_generate_sources": True,
                 "execute_steps": [
                     "dbt run-operation clone_all_production_schemas",