Hi @Jing1018,
I use the following code to be able to start a pipeline from the API and then get the status of the pipeline run:
# --- Pipeline start + status tracking ----------------------------------------
import time, json
from sempy.fabric import FabricRestClient
_client = FabricRestClient()
# Polling tunables (you can override these globals elsewhere)
FIRST_POLL_DELAY_SEC = 8 # initial delay before first poll
POLL_INTERVAL_SEC = 10 # steady-state poll interval
POLL_TIMEOUT_SEC = 60 * 60 # overall timeout (seconds)
def _start_job_instance_raw(ws_id: str, item_id: str, job_type: str, parameters: dict | None = None):
url = f"v1/workspaces/{ws_id}/items/{item_id}/jobs/instances?jobType={job_type}"
body = {}
if parameters is not None:
body = {"executionData": {"parameters": {k: {"value": str(v), "type": "string"} for k, v in parameters.items()}}}
print(f"[POST] {url} (paramKeys={list((parameters or {}).keys())})")
resp = _client.post(url, json=body)
print(f"[POST] → {resp.status_code} Location={resp.headers.get('Location')}")
return resp
def _start_job_instance(ws_id: str, item_id: str, job_type: str, parameters: dict | None = None) -> str:
"""Start a Fabric job and return the jobInstanceId; raises on non-202."""
resp = _start_job_instance_raw(ws_id, item_id, job_type, parameters)
if resp.status_code != 202:
try:
details = resp.json()
except Exception:
details = resp.text
raise RuntimeError(f"Start job failed ({resp.status_code}) jobType={job_type}: {details}")
loc = (resp.headers.get("Location") or "")
if "/jobs/instances/" not in loc:
raise RuntimeError(f"202 Accepted but missing job instance Location (jobType={job_type}).")
return loc.rsplit("/jobs/instances/", 1)[1]
def _get_job_instance(ws_id: str, item_id: str, job_instance_id: str) -> dict:
"""Fetch current state for a job instance."""
url = f"v1/workspaces/{ws_id}/items/{item_id}/jobs/instances/{job_instance_id}"
r = _client.get(url)
r.raise_for_status()
return r.json()
def start_pipeline_and_track(ws_id: str, item_id: str) -> tuple[int, str, dict]:
"""
Start a Pipeline job and wait for terminal status.
Returns: (rc, note_json, final_state_dict)
rc = 0 if completed, else 1
note_json = compact JSON summary (jobType, jobInstanceId, status, failureReason)
final_state_dict = full GET response from the last poll
"""
# 1) Start
job_instance_id = _start_job_instance(ws_id, item_id, "Pipeline", parameters=None)
# 2) Poll
print(f"[WARMUP] Sleeping {FIRST_POLL_DELAY_SEC}s before first poll (Pipeline {job_instance_id})")
time.sleep(FIRST_POLL_DELAY_SEC)
start_ts = time.time()
tries = 0
while True:
st = _get_job_instance(ws_id, item_id, job_instance_id)
status = (st.get("status") or "").lower()
print(f"[POLL] {job_instance_id} status={status}")
if status in ("completed", "failed", "cancelled", "canceled", "deduped"):
note = {
"jobType": "Pipeline",
"jobInstanceId": job_instance_id,
"status": status,
"failureReason": st.get("failureReason"),
}
return (0 if status == "completed" else 1), json.dumps(note, default=str), st
# backoff is simple here; you can make it smarter if you like
time.sleep(POLL_INTERVAL_SEC if tries < 3 else min(POLL_INTERVAL_SEC * 2, 30))
tries += 1
if (time.time() - start_ts) > POLL_TIMEOUT_SEC:
raise TimeoutError(f"Timeout waiting for Pipeline job {job_instance_id}; last state={st}")
If you found this helpful, consider giving Kudos. If I solved your problem or answered your question, mark this post as the solution.
