dbt job pipeline activity: bug

Since I wasn't able to wait for this bug to be resolved. I worked around the problem by scheduling the dbt job item via the API and then polling until complete in a notebook.

The full notebook is too large to share here, but here is a basic example of how it works. Note that some of this leverages the API in a way that is not documented yet, but it appears to work fine.

import time
from urllib.parse import unquote, urlparse

import requests

FABRIC_API_BASE_URL = "https://api.fabric.microsoft.com/v1"
WORKSPACE_ID = "<workspace-id>"
DBT_ITEM_ID = "<dbt-project-item-id>"
ACCESS_TOKEN = "<bearer-token>"  # acquire via your identity provider (e.g. MSAL / service principal)

# The command to execute: dbt operation + arguments (mirrors `dbt run --select ...`)
execution_data = {
    "operation": "run",
    "arguments": {
        "select": "tag:daily",
        "fullRefresh": False,
        "threads": 4,
    },
}

headers = {
    "Authorization": f"Bearer {ACCESS_TOKEN}",
    "Content-Type": "application/json",
}

# 1. Submit the job
submit_url = f"{FABRIC_API_BASE_URL}/workspaces/{WORKSPACE_ID}/items/{DBT_ITEM_ID}/jobs/Execute/instances"
response = requests.post(submit_url, headers=headers, json={"executionData": execution_data})
response.raise_for_status()

# Fabric returns the job instance location (may be relative or absolute)
location = response.headers["Location"]
job_instance_id = unquote(urlparse(location).path.rstrip("/").split("/")[-1])

# 2. Poll the job instance until it reaches a terminal state
instance_url = f"{FABRIC_API_BASE_URL}/workspaces/{WORKSPACE_ID}/items/{DBT_ITEM_ID}/jobs/instances/{job_instance_id}"

while True:
    status_response = requests.get(instance_url, headers=headers)
    status_response.raise_for_status()
    status = status_response.json().get("status", "").upper()

    if status in {"COMPLETED", "SUCCEEDED"}:
        print("dbt job succeeded")
        break
    if status in {"FAILED", "CANCELLED"}:
        raise RuntimeError(f"dbt job failed with status: {status}")

    time.sleep(20)  # poll interval