pipeline trigger using metadata table

Hi DiKi-I, 

This is going ot be a bit of work, but it is 100% doable. This is how things are run in my environment. 

At a high level, I have a lakehouse table that contains the details of the pipelines.  The important information to store is the ID of the workspace the pipeline is in, and the ID of the pipeline itself. In my environment, I am also storing some other information like the item type (my process can run notebooks as well as pipelines), the schedule I want it to run at, and dependencies that need to be completed before the item runs. 

Here are the columns I have in my job registry table: 

After you have this, you then need a notebook that can read the data and using the REST APIs start each pipeline. My notebook to do this is closer to 800 lines long, but it handles dependencies, scheduling, metadata collection, and a few other things. 

Here's the important function however, this is the function that uses the REST API to actually start the job. 

import sempy.fabric as fabric
from sempy.fabric import FabricRestClient
_client = FabricRestClient()

def _start_job_instance_raw(ws_id: str, item_id: str, job_type: str, parameters: Optional[Dict[str, Any]] = 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: Optional[Dict[str, Any]] = None) -> str:
    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 _handler_pipeline(job: Dict[str, Any]) -> Tuple[int, str]:
    ws_id   = job.get("workspace_id")
    item_id = job.get("item_id")
    if not ws_id or not item_id:
        return 1, "Missing workspace_id/item_id for pipeline run."
    dbg(f"[RUN] PIPELINE ws={ws_id} item={item_id}")
    ji = _start_job_instance(ws_id, item_id, "Pipeline", parameters=None)
    final = _poll_job_instance_with_warmup(ws_id, item_id, ji, job_type="Pipeline")
    st = (final.get("status") or "").lower()
    note = json.dumps({"jobType":"Pipeline","jobInstanceId":ji,"status":st,"failureReason":final.get("failureReason")}, default=str)
    dbg(f"[DONE] PIPELINE status={st}")
    return (0 if st=="completed" else 1), note

And here is my function that checks for the status of the job after it's been started: 

def _poll_job_instance_with_warmup(ws_id: str, item_id: str, job_instance_id: str, job_type: str = "RunNotebook") -> dict:
    start = time.time()
    tries = 0
    dbg(f"[WARMUP] Sleeping {FIRST_POLL_DELAY_SEC}s before first poll (jobType={job_type}, jobInstanceId={job_instance_id})")
    time.sleep(FIRST_POLL_DELAY_SEC)

    while True:
        st = _get_job_instance(ws_id, item_id, job_instance_id)
        status = (st.get("status") or "").lower()
        dbg(f"[POLL] {job_instance_id} status={status}")

        if status in ("completed", "failed", "cancelled", "canceled", "deduped"):
            if (job_type.lower() in ("runnotebook","notebook")
                and status == "failed"
                and _is_exec_state_not_found(st)
                and (time.time() - start) < EXEC_STATE_WARMUP_SEC):
                backoff = min(POLL_INTERVAL_SEC * max(2, tries + 1), 30)
                dbg(f"[POLL] transient NotFound during warm-up; retrying in {backoff}s")
                time.sleep(backoff); tries += 1; continue
            return st

        backoff = POLL_INTERVAL_SEC if tries < 3 else min(POLL_INTERVAL_SEC * 2, 30)
        time.sleep(backoff); tries += 1
        if time.time() - start > POLL_TIMEOUT_SEC:
            raise TimeoutError(f"Timeout waiting for job {job_instance_id}; last state={st}")

Note that these are just building blocks to give you an idea of how to achieve your goal, these snippets are not intended to be a full solution. 

If you found this helpful, consider giving some kudos. If I answered your quesiton or solved your problem, mark this post as the solution so future community members can find it easily. 

Thank you so much for your help. I was able to trigger the pipeline through metadata table using above functions. Through this notebook how you are passing this parameter to pipeline? How the pipeline should be configured?

also this solution will work across different workspaces? 

Hi DiKi-I ,

I wanted to check if you had the opportunity to review the information provided. Please feel free to contact us if you have any further questions.

Thank you.

Hi DiKi-I 

May I check if this issue has been resolved? If not, Please feel free to contact us if you have any further questions.

Thank you

Hi DiKi-I 

As we haven’t heard back from you, we wanted to kindly follow up to check if the suggestions  provided by the community members for the issue worked. Please feel free to contact us if you have any further questions.

Thanks and regards