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

resolves #1360

### description

`WorkflowJobApi.create()` passes plain dicts to
`workspace_client.jobs.create(**converted_job_spec)`. the databricks
sdk's `jobs.create()` iterates tasks calling `v.as_dict()` on each
(confirmed via `JobsAPI.create` source inspection). plain dicts have no
`.as_dict()` which caused the `'dict' object has no attribute 'as_dict'`
error.

1. the fix uses `JobSettings.from_dict()` + `.as_shallow_dict()` to
convert the plain dict job spec into proper sdk dataclasses before
calling `jobs.create()`.
2. `as_shallow_dict()` is the correct choice here : it unpacks top-level
fields as kwargs (matching the `jobs.create()` signature) while keeping
nested values as typed `Task` / `NotebookTask` / etc. dataclasses that
satisfy `.as_dict()`.
3. this is the same pattern already used by `update_job_settings()` in
the same class.

**changes:**
- `dbt/adapters/databricks/api_client.py` - `WorkflowJobApi.create()`
now deserializes the dict via `JobSettings.from_dict()` and passes
`.as_shallow_dict()` to the sdk. inline comment documents the sdk
internals that make this the correct choice.
- `tests/unit/api_client/test_workflow_job_api.py` - tests strengthened
to assert `isinstance(task, Task)`, call `.as_dict()` on each task and
verify output (directly proving the original `AttributeError` cannot
recur), and add a new `test_create__invalid_job_spec_raises` covering
the `JobSettings.from_dict()` error path.

### checklist

- [x] i have run this code in development and it appears to resolve the
stated issue
- [x] this pr includes tests, or tests are not required/relevant for
this pr
- [x] i have updated the `CHANGELOG.md` and added information about my
change to the "dbt-databricks next" section.

---------

Signed-off-by: aarushisingh04 <aarushi07.singh@gmail.com>
Co-authored-by: tejassp-db <241722411+tejassp-db@users.noreply.github.com>
Co-authored-by: Shubham Dhal <shubham.dhal@databricks.com>

Author: 3 people

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")