Forum Discussion
Jessy_D
10 months agoHelper I
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...
- 10 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 FabricRestClientdef 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.
tayloramy
10 months agoSuper User
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.
Jessy_D
9 months agoHelper I
Thanks, it was indeed intented to become a job, but this solved my issues