Merge pull request #192 from dbt-labs/fix/optional-schedule-for-merge-ci

feat: make schedule optional for ci and merge jobs

Author: b-per

src/dbt_jobs_as_code/schemas/job.py

@@ -20,6 +20,8 @@
 )
 from dbt_jobs_as_code.schemas.custom_environment_variable import CustomEnvironmentVariable
 
+JOB_TYPES_WITHOUT_SCHEDULE = ["ci", "merge"]
+
 
 @dataclass
 class IdentifierInfo:
@@ -57,10 +59,22 @@ def filter_jobs_by_import_filter(
     ]
 
 
+def _job_definition_json_schema_extra(schema: dict) -> None:
+    """Make 'schedule' conditionally required based on job_type."""
+    schema["if"] = {
+        "properties": {"job_type": {"enum": JOB_TYPES_WITHOUT_SCHEDULE}},
+        "required": ["job_type"],
+    }
+    schema["then"] = {}
+    schema["else"] = {"required": ["schedule"]}
+
+
 # Main model for loader
 class JobDefinition(BaseModel):
     """A definition for a dbt Cloud job."""
 
+    model_config = ConfigDict(json_schema_extra=_job_definition_json_schema_extra)
+
     linked_id: Optional[int] = Field(
         default=None,
         description="The ID of the job in dbt Cloud that we want to link. Only used for the 'link' command.",
@@ -85,7 +99,7 @@ class JobDefinition(BaseModel):
     errors_on_lint_failure: Optional[bool] = True
     execute_steps: List[str]
     generate_docs: bool
-    schedule: Schedule
+    schedule: Optional[Schedule] = None
     triggers: Triggers
     description: str = ""
     state: int = 1
@@ -217,6 +231,16 @@ def to_url(self, account_url: str) -> str:
         """Generate a URL for the job in dbt Cloud."""
         return f"{account_url}/deploy/{self.account_id}/projects/{self.project_id}/jobs/{self.id}"
 
+    @model_validator(mode="after")
+    def default_schedule_for_ci_merge(self):
+        """Default schedule for CI and merge jobs, require it for other job types."""
+        if self.schedule is None:
+            if self.job_type in JOB_TYPES_WITHOUT_SCHEDULE:
+                self.schedule = Schedule(cron="0 0 1 1 *")
+            else:
+                raise ValueError(f"'schedule' is required for '{self.job_type}' jobs")
+        return self
+
     @model_validator(mode="after")
     def validate_cron_expression(self):
         """Validate the cron expression and include job ID in error message if invalid."""

src/dbt_jobs_as_code/schemas/load_job_schema.json

@@ -100,6 +100,24 @@
     },
     "JobDefinition": {
       "description": "A definition for a dbt Cloud job.",
+      "else": {
+        "required": [
+          "schedule"
+        ]
+      },
+      "if": {
+        "properties": {
+          "job_type": {
+            "enum": [
+              "ci",
+              "merge"
+            ]
+          }
+        },
+        "required": [
+          "job_type"
+        ]
+      },
       "properties": {
         "linked_id": {
           "anyOf": [
@@ -252,7 +270,7 @@
               "type": "null"
             }
           ],
-          "default": false,
+          "default": true,
           "title": "Errors On Lint Failure"
         },
         "execute_steps": {
@@ -267,7 +285,15 @@
           "type": "boolean"
         },
         "schedule": {
-          "$ref": "#/$defs/Schedule"
+          "anyOf": [
+            {
+              "$ref": "#/$defs/Schedule"
+            },
+            {
+              "type": "null"
+            }
+          ],
+          "default": null
         },
         "triggers": {
           "$ref": "#/$defs/Triggers"
@@ -354,9 +380,9 @@
         "run_generate_sources",
         "execute_steps",
         "generate_docs",
-        "schedule",
         "triggers"
       ],
+      "then": {},
       "title": "JobDefinition",
       "type": "object"
     },

tests/schemas/test_job.py

@@ -1,5 +1,11 @@
+import json
+
 import pytest
+from jsonschema import ValidationError as JsonSchemaValidationError
+from jsonschema import validate
+from pydantic import ValidationError
 
+from dbt_jobs_as_code.schemas.config import generate_config_schema
 from dbt_jobs_as_code.schemas.job import (
     IdentifierInfo,
     JobDefinition,
@@ -153,3 +159,83 @@ def test_empty_filter(self, test_job_factory):
         assert len(result) == 1
         assert jobs[0] in result
         assert jobs[1] not in result
+
+
+BASE_JOB_DATA = {
+    "name": "Test Job",
+    "account_id": 1,
+    "project_id": 1,
+    "environment_id": 1,
+    "settings": {},
+    "triggers": {},
+    "execute_steps": ["dbt build"],
+    "run_generate_sources": False,
+    "generate_docs": False,
+}
+
+
+class TestScheduleConditionalRequirement:
+    """Tests for schedule being optional on ci/merge jobs."""
+
+    # -- Pydantic model tests --
+
+    @pytest.mark.parametrize("job_type", ["ci", "merge"])
+    def test_pydantic_schedule_optional_for_ci_merge(self, job_type):
+        job = JobDefinition(**{**BASE_JOB_DATA, "job_type": job_type})
+        assert job.schedule is not None  # defaults to Schedule()
+
+    @pytest.mark.parametrize("job_type", ["scheduled", "other"])
+    def test_pydantic_schedule_required_for_scheduled_other(self, job_type):
+        with pytest.raises(ValidationError, match="schedule"):
+            JobDefinition(**{**BASE_JOB_DATA, "job_type": job_type})
+
+    def test_pydantic_schedule_required_when_job_type_absent(self):
+        with pytest.raises(ValidationError, match="schedule"):
+            JobDefinition(**BASE_JOB_DATA)  # job_type defaults to "scheduled"
+
+    # -- JSON schema tests --
+
+    @pytest.fixture
+    def json_schema(self):
+        return json.loads(generate_config_schema())
+
+    def _config_instance(self, job_type=None, include_schedule=False):
+        """Build a minimal config dict for JSON schema validation."""
+        job = {
+            "name": "Test Job",
+            "account_id": 1,
+            "project_id": 1,
+            "environment_id": 1,
+            "settings": {},
+            "triggers": {},
+            "execute_steps": ["dbt build"],
+            "run_generate_sources": False,
+            "generate_docs": False,
+        }
+        if job_type is not None:
+            job["job_type"] = job_type
+        if include_schedule:
+            job["schedule"] = {"cron": "0 0 * * *"}
+        return {"jobs": {"test_job": job}}
+
+    @pytest.mark.parametrize("job_type", ["ci", "merge"])
+    def test_json_schema_schedule_optional_for_ci_merge(self, json_schema, job_type):
+        instance = self._config_instance(job_type=job_type, include_schedule=False)
+        validate(instance=instance, schema=json_schema)
+
+    @pytest.mark.parametrize("job_type", ["scheduled", "other"])
+    def test_json_schema_schedule_required_for_scheduled_other(self, json_schema, job_type):
+        instance = self._config_instance(job_type=job_type, include_schedule=False)
+        with pytest.raises(JsonSchemaValidationError, match="schedule"):
+            validate(instance=instance, schema=json_schema)
+
+    def test_json_schema_schedule_required_when_job_type_absent(self, json_schema):
+        instance = self._config_instance(include_schedule=False)
+        with pytest.raises(JsonSchemaValidationError, match="schedule"):
+            validate(instance=instance, schema=json_schema)
+
+    @pytest.mark.parametrize("job_type", ["scheduled", "ci", "merge", "other"])
+    def test_json_schema_schedule_accepted_for_all_types(self, json_schema, job_type):
+        """Providing schedule is always valid regardless of job_type."""
+        instance = self._config_instance(job_type=job_type, include_schedule=True)
+        validate(instance=instance, schema=json_schema)

tests/schemas/test_json_schema_up_to_date.py

@@ -0,0 +1,19 @@
+import json
+from pathlib import Path
+
+from dbt_jobs_as_code.schemas.config import generate_config_schema
+
+SCHEMA_PATH = Path("src/dbt_jobs_as_code/schemas/load_job_schema.json")
+
+
+def test_json_schema_is_up_to_date():
+    """Ensure the committed JSON schema matches what the models generate.
+
+    If this fails, run: `dbt-jobs-as-code update-json-schema`
+    """
+    committed = json.loads(SCHEMA_PATH.read_text())
+    generated = json.loads(generate_config_schema())
+    assert committed == generated, (
+        "The committed JSON schema is out of date. "
+        "Run `dbt-jobs-as-code update-json-schema` to regenerate it."
+    )