Forum Discussion

Jessy_D's avatar
Jessy_D
Helper I
9 months ago
Solved

Working with mutliple nested notebooks

I have a Pyspark (python) notebook that imports other notebooks. When the notebooks are in the same folder, there is no issue when I use the %run command to import the notebooks, but when I place som...
  • tayloramy's avatar
    9 months ago

    Hi Jessy_D

     

    In addition to what ibarrau mentioned, you can also use the Scheduler API. Here's my code snippet to do that: 
    self.client is a FabricRestClient: 

    from sempy.fabric import FabricRestClient
     
     def start_job_instance(
            self,
            ws_id: str,
            item_id: str,
            job_type: str,
            parameters: Optional[Dict[str, Any]] = None
        ) -> str:
            """
            Start a Fabric job instance and return the job instance ID.
    
            Args:
                ws_id: Workspace GUID
                item_id: Item GUID (notebook or pipeline)
                job_type: "RunNotebook", "Notebook", or "Pipeline"
                parameters: Optional dict of parameter name -> value
    
            Returns:
                Job instance ID (GUID)
    
            Raises:
                RuntimeError: If job start fails
            """
            url = f"v1/workspaces/{ws_id}/items/{item_id}/jobs/instances?jobType={job_type}"
    
            # Build request body
            body = {}
            if parameters:
                body = {
                    "executionData": {
                        "parameters": {
                            k: {"value": str(v), "type": "string"}
                            for k, v in parameters.items()
                        }
                    }
                }
    
            log_debug(f"POST {url} params={list((parameters or {}).keys())}")
            resp = self.client.post(url, json=body)
            log_debug(f"POST response: {resp.status_code} Location={resp.headers.get('Location')}")
    
            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}"
                )
    
            # Extract job instance ID from Location header
            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]

     

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