Integration with DAB workflows

[kyrre]

I want to use the SDK to trigger a workflows defined using DAB (in the same project), e.g., the layout is as follows:

├── README.md
├── [service_principal_name].parquet.py
├── azure_applications
│   ├── __init__.py
│   ├── task.py
│   └── test_task.py
├── conftest.py
├── databricks.yml
├── dist
│   └── azure_applications-0.0.3-py3-none-any.whl
├── pyproject.toml
├── resources
│   └── azure_applications_job.yml
└── uv.lock

where the file [service_principal_name].parquet.py should run the job:

w = WorkspaceClient(profile="security")

name = "[dev kyrre_wahl_kongsgard] azure_applications_job")
job_id = w.jobs.list(limit=1, expand_tasks=True, name=name)[0].job_id

run = w.jobs.run_now_and_wait(
    job_id=job_id,
    python_named_params={
       ...
    },
)

is there any way do this in a seamless way so I easily switch between the different targets without manually specifying the target name? 🤔

[mateuscarestiato]

Hi @kyrre! Your current approach (searching by name) works but has a few fragilities. Here are the recommended patterns for triggering DAB-managed workflows via the Python SDK:

Your current approach — improved:

from databricks.sdk import WorkspaceClient

w = WorkspaceClient(profile="security")

Use next() with a generator for safety (avoids IndexError if job not found)

job_name = "[dev kyrre_wahl_kongsgard] azure_applications_job"

job = next(
(j for j in w.jobs.list(expand_tasks=False) if j.settings.name == job_name),
None
)

if job is None:
raise ValueError(f"Job '{job_name}' not found")

run = w.jobs.run_now_and_wait(
job_id=job.job_id,
python_named_params={"param1": "value1"}
)
print(f"Run completed with state: {run.state.result_state}")

Better approach — use the job_id from DAB outputs:

After databricks bundle deploy, the bundle stores the created job IDs in the bundle state. You can retrieve them:

Get job ID from bundle metadata

databricks bundle summary --output json | jq '.resources.jobs.<job_key>.id'

Then hardcode or inject that ID into your SDK call — avoids the name search entirely.

Recommended pattern for CI/CD integration:

import os
from databricks.sdk import WorkspaceClient

w = WorkspaceClient(profile="security")

Inject job_id from environment (set by your CI/CD pipeline from DAB outputs)

job_id = int(os.environ["MY_JOB_ID"])

run = w.jobs.run_now_and_wait(
job_id=job_id,
python_named_params={"env": "dev", "date": "2025-01-01"},
)

For listing with pagination (important for large workspaces):

w.jobs.list() returns a generator — works fine, but add a name filter for speed

jobs = w.jobs.list(name=job_name) # server-side filter, much faster
job = next(jobs, None)

Note: w.jobs.list(limit=1, name=job_name) with exact name match is actually the fastest approach — no need to iterate all jobs. The name parameter in the API does server-side filtering.

Let me know if you're triggering this from a CI pipeline or from another Databricks notebook!