fix: convert workflow job spec to SDK dataclasses before jobs.create() (#1338)

CHANGELOG.md

@@ -6,6 +6,7 @@
 - 
 ### Fixes
 
+- Fix `workflow_job` Python model submission method failing with dictionary attribute error ([#1360](https://github.com/databricks/dbt-databricks/issues/1360))
 - Fix column order mismatch in microbatch and replace_where incremental strategies by using INSERT BY NAME syntax ([#1338](https://github.com/databricks/dbt-databricks/issues/1338))
 
 ## dbt-databricks 1.11.6 (Mar 10, 2026)

dbt/adapters/databricks/api_client.py

@@ -801,7 +801,10 @@ def create(self, job_spec: dict[str, Any]) -> str:
         try:
             # Convert job_spec to be compatible with jobs.create
             converted_job_spec = self._convert_job_spec_for_create(job_spec)
-            create_response = self.workspace_client.jobs.create(**converted_job_spec)
+            # Convert plain dicts to SDK dataclasses via JobSettings to avoid
+            # AttributeError when the SDK iterates and calls .as_dict() on tasks.
+            job_settings = JobSettings.from_dict(converted_job_spec)
+            create_response = self.workspace_client.jobs.create(**job_settings.as_shallow_dict())
             job_id = str(create_response.job_id)
             logger.info(f"New workflow created with job id {job_id}")
             return job_id

tests/unit/api_client/test_workflow_job_api.py

@@ -1,7 +1,7 @@
-from unittest.mock import Mock
+from unittest.mock import Mock, patch
 
 import pytest
-from databricks.sdk.service.jobs import BaseJob, CreateResponse, JobSettings, QueueSettings
+from databricks.sdk.service.jobs import BaseJob, CreateResponse, JobSettings, QueueSettings, Task
 from dbt_common.exceptions import DbtRuntimeError
 
 from dbt.adapters.databricks.api_client import WorkflowJobApi
@@ -59,7 +59,10 @@ def test_create__success(self, api, workspace_client):
         result = api.create(job_spec)
 
         assert result == "456"
-        workspace_client.jobs.create.assert_called_once_with(**job_spec)
+        workspace_client.jobs.create.assert_called_once()
+        call_kwargs = workspace_client.jobs.create.call_args[1]
+        assert call_kwargs["name"] == "test_job"
+        assert call_kwargs.get("tasks") in (None, [])
 
     def test_create__job_spec_conversion(self, api, workspace_client):
         mock_create_response = Mock(spec=CreateResponse)
@@ -89,21 +92,42 @@ def test_create__job_spec_conversion(self, api, workspace_client):
         assert result == "789"
         workspace_client.jobs.create.assert_called_once()
 
-        # Verify the call was made with converted parameters
+        # Tasks are converted to SDK Task dataclasses via JobSettings.from_dict().as_shallow_dict().
+        # The SDK's jobs.create() calls v.as_dict() on each task internally, so
+        # we verify: (a) they are Task instances, (b) attributes are correct,
+        # (c) .as_dict() works (proving the original AttributeError is fixed).
         call_kwargs = workspace_client.jobs.create.call_args[1]
         assert call_kwargs["name"] == "test_job"
-        assert len(call_kwargs["tasks"]) == 2
-
-        # Check first task conversion
-        task1 = call_kwargs["tasks"][0]
-        assert task1["task_key"] == "task1"
-        assert "cluster_id" not in task1  # Should be removed
-        assert task1["existing_cluster_id"] == "test-cluster-id"  # Should be converted
-
-        # Check second task remains unchanged
-        task2 = call_kwargs["tasks"][1]
-        assert task2["task_key"] == "task2"
-        assert task2["existing_cluster_id"] == "already-correct"
+        tasks = call_kwargs["tasks"]
+        assert len(tasks) == 2
+
+        task1 = tasks[0]
+        assert isinstance(task1, Task)
+        assert task1.task_key == "task1"
+        assert task1.existing_cluster_id == "test-cluster-id"
+        # Verify .as_dict() works — this was the root cause of the bug
+        task1_dict = task1.as_dict()
+        assert task1_dict["task_key"] == "task1"
+        assert task1_dict["existing_cluster_id"] == "test-cluster-id"
+
+        task2 = tasks[1]
+        assert isinstance(task2, Task)
+        assert task2.task_key == "task2"
+        assert task2.existing_cluster_id == "already-correct"
+        assert task2.as_dict()["existing_cluster_id"] == "already-correct"
+
+    def test_create__invalid_job_spec_raises(self, api, workspace_client):
+        """If JobSettings.from_dict() raises (e.g. on malformed input), create()
+        wraps it in a DbtRuntimeError via the existing except block."""
+        with patch(
+            "dbt.adapters.databricks.api_client.JobSettings.from_dict",
+            side_effect=Exception("malformed spec"),
+        ):
+            with pytest.raises(DbtRuntimeError) as exc_info:
+                api.create({"name": "bad_job", "tasks": ["not-a-dict"]})
+
+        assert "Error creating Workflow" in str(exc_info.value)
+        workspace_client.jobs.create.assert_not_called()
 
     def test_update_job_settings__exception(self, api, workspace_client):
         workspace_client.jobs.reset.side_effect = Exception("API Error")