Running Fabric Pipelines from a Notebook

Everyone knows the easy direction: drop a Notebook activity into a pipeline, and the pipeline waits for it to finish before moving on. The reverse, a notebook kicking off a pipeline and waiting for success or failure, is less documented.

Here’s a working pattern for that: trigger the pipeline you want to run, monitor it at the activity level rather than just the top-level job status, and raise on failure.

For the rest of this post, treat workspace_id and pipeline_id as placeholders (fill your own values). The REST API needs the item’s GUID rather than its display name, so these are just the two values you copy out of the pipeline’s URL in the Fabric portal and paste in.

Monitor at the activity level, not just the job status

The Job Scheduler API gives you a top-level status for the job instance itself, but that alone doesn’t tell you what happened inside the pipeline. A more reliable read is to poll the pipeline’s activity runs and treat the job as done once every activity has reached a terminal state:

from datetime import datetime, timedelta
import time

def get_activity_runs(workspace_id: str, pipeline_run_id: str):
    url = f"/v1/workspaces/{workspace_id}/datapipelines/pipelineruns/{pipeline_run_id}/queryactivityruns"

    now = datetime.utcnow()
    payload = {
        "filters": [],
        "orderBy": [{"orderBy": "ActivityRunStart", "order": "DESC"}],
        "lastUpdatedAfter":  (now - timedelta(hours=2)).strftime("%Y-%m-%dT%H:%M:%SZ"),
        "lastUpdatedBefore": (now + timedelta(hours=2)).strftime("%Y-%m-%dT%H:%M:%SZ"),
    }

    resp = client.post(url, json=payload)
    resp.raise_for_status()
    data = resp.json()

    if isinstance(data, dict) and "value" in data:
        return data["value"]
    return data

def wait_for_pipeline_via_activities(workspace_id, pipeline_run_id, poll_interval=10, timeout=3600):
    start = time.time()
    while True:
        activities = get_activity_runs(workspace_id, pipeline_run_id)

        if activities:
            statuses = {a["activityName"]: a["status"] for a in activities}

            if any(s not in ("Succeeded", "Failed", "Cancelled") for s in statuses.values()):
                pass  # keep polling
            else:
                if any(s != "Succeeded" for s in statuses.values()):
                    return "Failed", activities
                return "Succeeded", activities
        else:
            # no activities yet – job may be queued / starting
            pass

        if time.time() - start > timeout:
            raise TimeoutError(f"Timed out waiting for run {pipeline_run_id}")

        time.sleep(poll_interval)

def get_effective_pipeline_status(workspace_id: str, pipeline_run_id: str) -> str:
    activities = get_activity_runs(workspace_id, pipeline_run_id)
    if not activities:
        return "Unknown"

    if any(a.get("status") != "Succeeded" for a in activities):
        return "Failed"
    return "Succeeded"

The lastUpdatedAfter/lastUpdatedBefore window (two hours either side of “now”) is just a filter on the activity-run query, widen it if you expect a pipeline to run longer than that.

Trigger the pipeline and wait

sempy’s FabricRestClient authenticates as the signed-in user, so there’s no token handling to write:

try:
    response = client.post(relative_url, json={})
except Exception as e
	error_message = str(e):
    raise Exception(f"Failed to run the pipeline: {pipeline_id}")

# monitor the pipeline run until completion
pipeline_run_id = response.headers["Location"].split("instances/")[1]

# Wait for every activity to finish, then check the outcome
wait_for_pipeline_via_activities(workspace_id, pipeline_run_id)
status = get_effective_pipeline_status(workspace_id, pipeline_run_id)

# fail the notebook
if status == "Failed":
    raise Exception(f"Pipeline {pipeline_id} with run ID {pipeline_run_id} failed")

The takeaway

A pipeline run is an async job under the hood: trigger it, poll it, and decide what “done” means for your own pipelines. Here, that’s every activity reaching a terminal status rather than just the top-level job finishing. Once that’s in place, a pipeline slots into a notebook-driven orchestration flow the same way a child notebook would.

Tags: Fabric, Notebooks, Python