Hi @v-shex-msft
Initially, I generate an access token using the Client ID, Client Secret, and Tenant ID, while specifying the necessary scopes (Workspace.ReadWrite.All, Item.ReadWrite.All). With the freshly obtained token, I proceed to call the Get Definition API using the Workspace ID and Notebook ID.
Upon invoking the API, I receive a 202 Accepted status code, indicating the request is being processed. To handle this, I implemented a polling mechanism that repeatedly checks the Location URL provided in the response header. This polling system ensures I can track the progress of the request asynchronously.
However, when navigating to the Location URL, I encounter a 403 Forbidden error. To address this, I have also incorporated an async/await system to fetch the link and concurrently retrieve the latest status in real-time.
so below i am attaching 2 approaches i used to tackle this situation can you please give your insights on this
1) without asynch await
import requests
import time
import json
workspaceID = ''
notebookId = ''
api_url = f'https://api.fabric.microsoft.com/v1/workspaces/{workspaceID}/notebooks/{notebookId}/getDefinition'
access_token = ''
headers = {
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json"
}
response = requests.post(api_url, headers=headers)
if response.status_code == 200:
notebooks = response.json()
print("Response from the API:")
print(json.dumps(notebooks, indent=4))
elif response.status_code == 202:
print("Request accepted. Processing is not yet complete.")
location_url = response.headers.get('Location')
if location_url:
print(f"Check the status at: {location_url}")
while True:
time.sleep(5)
status_response = requests.get(location_url, headers=headers)
if status_response.status_code == 200:
result = status_response.json()
print("Final result:")
print(json.dumps(result, indent=4))
break
elif status_response.status_code == 202:
print("Still processing...")
else:
print(f"Error checking status: {status_response.status_code}, {status_response.text}")
break
else:
print("No location URL provided for status checking.")
else:
print(f"Failed to fetch notebooks: {response.status_code}, {response.text}")
2) With Async await
import aiohttp
import asyncio
import json
import nest_asyncio
nest_asyncio.apply()
workspaceID = ''
notebookId = ''
api_url = f'https://api.fabric.microsoft.com/v1/workspaces/{workspaceID}/notebooks/{notebookId}/getDefinition'
access_token = ''
async def fetch_notebook(session):
headers = {
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json"
}
async with session.post(api_url, headers=headers) as response:
if response.status == 200:
notebooks = await response.json()
print("Response from the API:")
print(json.dumps(notebooks, indent=4))
elif response.status == 202:
print("Request accepted. Processing is not yet complete.")
location_url = response.headers.get('Location')
if location_url:
await poll_status(session, location_url)
else:
print("No location URL provided for status checking.")
else:
print(f"Failed to fetch notebooks: {response.status}, {await response.text()}")
async def poll_status(session, location_url):
while True:
async with session.get(location_url) as status_response:
if status_response.status == 200:
final_result = await status_response.json()
print("Final result:")
print(json.dumps(final_result, indent=4))
break
elif status_response.status == 202:
print("Still processing...")
await asyncio.sleep(5)
else:
print(f"Error checking status: {status_response.status}, {await status_response.text()}")
break
async def main():
async with aiohttp.ClientSession() as session:
await fetch_notebook(session)
await main()
Please let me know if there are any changes to be made in the code
Thanks!!
