Introduction

When working with DLT pipelines, you want to retrieve DLT pipeline details, including the following. 

• Source code (notebooks or file paths)
• Configuration parameters
• Permissions associated with the pipeline

You want to retrieve pipeline details to be able to do the following. 

• Audit pipeline configurations.
• Export DLT definitions to CI/CD systems.
• Review access controls for governance.

Instructions

Adapt and run the following Python code in a notebook to retrieve DLT pipeline details. 

import requests
import logging

# === Configure logging ===
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s"
)

# === Set your environment variables ===
DATABRICKS_INSTANCE = "https://<databricks-host>.cloud.databricks.com"
DATABRICKS_TOKEN = "dapi<your-personal-access-token>"
PIPELINE_ID = "<your-dlt-pipeline-id>"

headers = {
    "Authorization": f"Bearer {DATABRICKS_TOKEN}"
}

# === Get pipeline info ===
pipeline_url = f"{DATABRICKS_INSTANCE}/api/2.0/pipelines/{PIPELINE_ID}"
response = requests.get(pipeline_url, headers=headers)
response.raise_for_status()
pipeline_info = response.json()

logging.info("Pipeline Name: %s", pipeline_info.get("name"))

logging.info("Source Code Libraries:")
for lib in pipeline_info["spec"].get("libraries", []):
    if "notebook" in lib:
        logging.info("  - Notebook: %s", lib["notebook"]["path"])
    elif "file" in lib:
        logging.info("  - File: %s", lib["file"]["path"])

logging.info("Configuration Parameters:")
for k, v in pipeline_info["spec"].get("configuration", {}).items():
    logging.info("  - %s: %s", k, v)

# === Get pipeline permissions ===
permissions_url = f"{DATABRICKS_INSTANCE}/api/2.0/permissions/pipelines/{PIPELINE_ID}"
perm_response = requests.get(permissions_url, headers=headers)
perm_response.raise_for_status()
permissions = perm_response.json()

logging.info("Pipeline Permissions:")
for acl in permissions.get("access_control_list", []):
    principal = (
        acl.get("user_name") or
        acl.get("group_name") or
        acl.get("service_principal_name")
    )
    perms = [p["permission_level"] for p in acl.get("all_permissions", [])]
    logging.info("  - %s: %s", principal, ", ".join(perms))