User Data Function coroutine object error

Hi amaaiia, 

The string "<coroutine object ...>" appears when an async function is called without await. You don’t need async here unless you truly want concurrency. Use a sync HTTP client and your wrapper functions will work as-is.

• Use requests or httpx in sync mode (both are fine). See HTTPX’s sync vs async note: httpx async support.

import httpx

def get_element_schedules(token: str, workspace_id: str, job_type: str) -> dict:
    headers = {"Authorization": f"Bearer {token}"}

    if job_type == "Pipeline":
        items_url = f"https://api.fabric.microsoft.com/v1/workspaces/{workspace_id}/datapipeline/items"
        schedule_job_type = "Pipeline"
    elif job_type == "Notebook":
        items_url = f"https://api.fabric.microsoft.com/v1/workspaces/{workspace_id}/notebook/items"
        schedule_job_type = "DefaultJob"  # notebooks use DefaultJob for schedules
    else:
        raise ValueError("job_type must be Pipeline or Notebook")

    items = httpx.get(items_url, headers=headers, timeout=60).json().get("value", [])
    result = []
    for item in items:
        item_id = item["id"]
        sched_url = (
            f"https://api.fabric.microsoft.com/v1/workspaces/{workspace_id}"
            f"/items/{item_id}/jobs/{schedule_job_type}/schedules"
        )
        schedules = httpx.get(sched_url, headers=headers, timeout=60).json().get("value", [])
        result.append({"itemId": item_id, "name": item.get("displayName"), "schedules": schedules})

    return {"result": result}

def get_pipelines_schedules(token: str, workspace_id: str) -> dict:
    return get_element_schedules(token, workspace_id, job_type="Pipeline")

def get_notebooks_schedules(token: str, workspace_id: str) -> dict:
    return get_element_schedules(token, workspace_id, job_type="Notebook")

If you found this helpful, consider giving some Kudos. If I answered your question or solved your problem, mark this post as the solution.