import requests
import base64
# Databricks Workspace API URLs
workspace_url = f"{host}/api/2.0/workspace"
export_url = f"{workspace_url}/export"
import_url = f"{workspace_url}/import"
# Databricks personal access token
token = "***********************"
# Notebook path
notebook_path = "/Workspace/Users/Avinash/Old_Notebook"
new_notebook_path = "/Workspace/Users/Avinash/New_notebook"
# Function to export notebook content
def export_notebook():
export_payload = {
"path": notebook_path,
"format": "SOURCE"
}
response = requests.get(export_url, headers=headers, json=export_payload)
response.raise_for_status()
return response.json()["content"]
# Function to refactor notebook content
def refactor_notebook(content๐
# Example: Replace all occurrences of 'old_variable' with 'new_variable'
new_content = content.replace("hive_metastore", "test_catalog_ph")
return new_content
# Function to import modified notebook content
def import_notebook(new_content):
import_payload = {
"path": new_notebook_path,
"content": new_content,
# "overwrite": True
}
response = requests.post(import_url, headers=headers, json=import_payload)
if response.status_code != 200:
print(response.content)
response.raise_for_status()
# Main script
if __name__ == "__main__":
# Export notebook content
notebook_content = export_notebook()
decoded_content = base64.b64decode(notebook_content).decode('utf-8')
# print(decoded_content)
# Refactor notebook content
new_notebook_content = refactor_notebook(decoded_content)
encoded_content=(base64.b64encode(new_notebook_content.encode('utf-8')))
#print(encoded_content)
#print(base64.b64decode(encoded_content).decode('utf-8'))
# Import modified content back to Databricks
import_notebook(encoded_content)
print("Notebook refactoring complete.")