update_notebook_default_lakehouse.py

import requests
import time
import json
import base64
 
def get_notebook_content(notebook_id_or_name):
    nb = notebookutils.notebook.get(notebook_id_or_name)
    workspaceId = nb['workspaceId']
    notebookId = nb['id']
    format = 'ipynb'
    headers = {
        "Authorization": "Bearer " + notebookutils.credentials.getToken("pbi")
    }
    url = f"https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/notebooks/{notebookId}/getDefinition?format={format}"
    response = requests.post(url, headers=headers)
 
    if response.status_code == 202:
        retry_after = int(response.headers.get("Retry-After", 5))
        location = response.headers.get("Location")
        for _ in range(3):
            time.sleep(retry_after)
            resp = requests.get(f"{location}/result", headers=headers)
            if resp.status_code == 200:
                return resp.text
    elif response.status_code == 200:
        return response.text
    raise Exception("get notebook context failed.")
 
def update_notebook_default_lakehouse(notebook_id_or_name, default_lakehouse, default_lakehouse_workspace_id):
    content = get_notebook_content(notebook_id_or_name)
    payload = json.loads(content)["definition"]["parts"][0]["payload"]
    content = json.loads(base64.b64decode(payload).decode('utf-8'))
    # remove current dependencies
    del content['metadata']['dependencies']
    # print(content)
    return notebookutils.notebook.updateDefinition(notebook_id_or_name, content, default_lakehouse, default_lakehouse_workspace_id)
 
notebook_id_or_name = "xxxxxx"
default_lakehouse = "LH2"
default_lakehouse_workspace_id = "b738ec83-f152-4600-9046-d069f2714e38"
resp = update_notebook_default_lakehouse(notebook_id_or_name, default_lakehouse, default_lakehouse_workspace_id)
print(resp)