如何使用Python调用REST API导出下载Azure Databricks笔记本到本地

这个需求完全可落地，核心调用Databricks官方Workspace REST API即可，不需要依赖额外的第三方SDK，具体实现如下：

前置准备 #

• 生成Azure Databricks个人访问令牌（PAT）：进入工作区用户设置页，找到开发者选项下的访问令牌生成入口，创建新令牌并保存令牌值，该值仅在创建时展示一次
• 记录工作区实例URL，格式为https://<你的工作区资源ID>.azuredatabricks.net
• 确认待导出notebook的工作区路径，例如/Users/your_account@domain.com/demo_nb；API支持的导出格式包括SOURCE（纯源码格式）、HTML（带运行结果的静态页面）、JUPYTER（标准.ipynb格式）、DBC（Databricks原生归档格式）

单文件导出实现 #

首先安装唯一依赖：
pip install requests

核心代码如下：

import os
import requests
import base64

# 配置参数建议从环境变量读取，不要硬编码在脚本里
DATABRICKS_INSTANCE = os.getenv("DB_INSTANCE", "替换为你的工作区URL")
DATABRICKS_TOKEN = os.getenv("DB_PAT", "替换为你的个人访问令牌")
TARGET_NB_PATH = "/替换为待导出notebook的工作区路径"
LOCAL_SAVE_DIR = "./exported_notebooks"
EXPORT_FORMAT = "SOURCE"

req_headers = {
    "Authorization": f"Bearer {DATABRICKS_TOKEN}",
    "Content-Type": "application/json"
}

def export_nb(workspace_nb_path, local_save_path, export_format="SOURCE"):
    export_api = f"{DATABRICKS_INSTANCE}/api/2.0/workspace/export"
    req_params = {
        "path": workspace_nb_path,
        "format": export_format,
        "direct_download": "false"
    }
    resp = requests.get(export_api, headers=req_headers, params=req_params)
    resp.raise_for_status()
    
    # API返回的文件内容为base64编码，需解码后写入本地
    nb_content = base64.b64decode(resp.json()["content"])
    # 按导出格式匹配文件后缀
    suffix_map = {
        "SOURCE": ".py",
        "HTML": ".html",
        "JUPYTER": ".ipynb",
        "DBC": ".dbc"
    }
    os.makedirs(os.path.dirname(local_save_path), exist_ok=True)
    full_save_path = f"{local_save_path}{suffix_map[export_format]}"
    with open(full_save_path, "wb") as f:
        f.write(nb_content)
    print(f"notebook已保存至: {full_save_path}")

if __name__ == "__main__":
    nb_file_name = os.path.basename(TARGET_NB_PATH)
    export_nb(TARGET_NB_PATH, os.path.join(LOCAL_SAVE_DIR, nb_file_name), EXPORT_FORMAT)

批量导出整个目录扩展 #

如果需要批量导出整个文件夹下的所有notebook，可以先调用/api/2.0/workspace/list接口遍历路径下的所有对象，区分notebook和子目录做递归导出即可，核心遍历逻辑参考：

def list_workspace_items(target_path):
    list_api = f"{DATABRICKS_INSTANCE}/api/2.0/workspace/list"
    resp = requests.get(list_api, headers=req_headers, params={"path": target_path})
    resp.raise_for_status()
    return resp.json().get("objects", [])

def export_dir(workspace_dir, local_root_dir):
    items = list_workspace_items(workspace_dir)
    for item in items:
        item_path = item["path"]
        if item["object_type"] == "NOTEBOOK":
            # 保持本地目录结构和工作区完全一致
            relative_path = os.path.relpath(item_path, workspace_dir)
            local_nb_path = os.path.join(local_root_dir, relative_path)
            export_nb(item_path, local_nb_path)
        elif item["object_type"] == "DIRECTORY":
            export_dir(item_path, local_root_dir)

自动化注意事项 #

• 生产环境不要把访问令牌硬编码在脚本中，建议通过环境变量、Azure Key Vault等密钥管理服务读取，避免令牌泄露
• 定时执行可以直接挂Linux crontab、Windows任务计划，或者接入CI/CD流水线按触发规则运行
• 导出SOURCE格式时，如果notebook使用SQL、Scala、R语言，可以通过list接口返回的language字段自动匹配对应文件后缀，不需要手动修改
• 如果需要保留单元格运行结果，选择HTML或JUPYTER格式导出即可，SOURCE格式仅保存代码逻辑不存储运行输出

内容的提问来源于stack exchange，提问作者PAVAN