How to check the status of a pipeline run submitted via API

[Jing1018]

Hi, I want to create a Python notebook to submit pipeline runs and check their status programmatically. Two APIs I plan to use are:
1. Run on demand pipeline job: https://community.fabric.microsoft.com/t5/Data-Pipeline/Execute-Data-Pipeline-Via-API/m-p/3740462 

2. Get pipeline job instance: https://learn.microsoft.com/en-us/fabric/data-factory/pipeline-rest-api-capabilities#get-pipeline-jo... 
I initially assumed the two APIs would integrate seamlessly. However, during implementation, I realized that the 'run on demand' API only returns a status code and not the job run ID. Is there a way to retrieve the IDs of jobs submitted through the 'run on demand' API?

[tayloramy]

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. 

[Jing1018]

After syncing internally with domain experts, we discovered that—despite the lack of mention in the official documentation—the API call does return some metadata. For example, the job run URL can be fetched via response.headers['location'].

Sample Response Headers:

cache-control: no-store, must-revalidate, no-cache

pragma: no-cache

content-type: application/octet-stream

location: https://msitapi.fabric.microsoft.com/v1/workspaces/79294ce5-364b-4936-b251-f5a864181ddd/items/d83195...

retry-after: 60

x-ms-job-id: 38fd5429-136b-4041-9353-211d561039f8

strict-transport-security: max-age=31536000; includeSubDomains

x-frame-options: deny

x-content-type-options: nosniff

requestid: 1372206e-f857-4e10-9a98-fbbcc5ce0c55

access-control-expose-headers: RequestId,Location,Retry-After

date: Wed, 27 Aug 2025 05:06:33 GMT

[v-prasare]

Hi @Jing1018, Thanks for actively updating the thread, you insights truely help our community users. please do post your progress here.

Thanks,

Prashanth Are

MS Fabric community support

[Jing1018]

Filtering by PipelineId and time range isn't a sufficiently reliable solution—especially since I plan to submit pipeline runs in batches. At this point, the only viable workaround seems to be implementing custom logging that captures an exhaustive list of parameter values along with the PipelineRunId and PipelineId. However, this feels unnecessarily complex for addressing such a fundamental need. I have proposed an idea at "Run on demand pipeline run" API returns job run I... - Microsoft Fabric Community

[v-prasare]

Thanks for submitting idea in ideas forum. Feedback submitted here is often reviewed by the product teams and can lead to meaningful improvement.

Thanks

Prashanth

MS Fabric community

[Anonymous]

Plz refer the below blog for trigger and status API integrations :

https://datasharkx.wordpress.com/2025/03/22/trigger-fabric-data-pipeline-from-azure-data-factory-syn...

[Vinodh247]

The run on Demand API currently returns only an HTTP status code, such as 202 Accepted, without providing the runId needed to directly query the status of the submitted pipeline run. To work around this limitation, you can use the Get Pipeline Jobs endpoint to retrieve a list of recent pipeline runs after triggering the job. By filtering the results using the pipelineID and the timestamp close to your submission time, you can identify the specific run and extract its runId. Once you have the runId, you can poll the Get Pipeline Job Instance API to check the run status and execution details, such as start time, end time, and current status. If you need a guaranteed way to link the pipeline trigger to its run, consider implementing a logging mechanism that records metadata during the trigger process or explore triggering through the Fabric SDK when it becomes available for more seamless integration.

Please 'Kudos' and 'Accept as Solution' if this answered your query.