databricks-isv-python-dbconnect

<!-- skill-version: 1.0.0 --> # Databricks Connect for Python (ISV) Use this skill when implementing or testing **Databricks Connect** (`databricks-connect`) integrations for PWAF-compliant Spark applications running on Databricks compute (serverless or classic clusters). ## PWAF Documentation Links | Resource | URL | |----------|-----| | **PWAF Home** | https://databrickslabs.github.io/partner-architecture/ | | **Authentication Best Practices** | https://databrickslabs.github.io/partner-architecture/isv-partners/lakehouse-patterns/access-auth/ | | **OAuth M2M Guide** | https://databrickslabs.github.io/partner-architecture/isv-partners/lakehouse-patterns/access-auth/oauth-m2m | | **OAuth U2M Guide** | https://databrickslabs.github.io/partner-architecture/isv-partners/lakehouse-patterns/access-auth/oauth-u2m | | **Databricks Connect Telemetry** | https://databrickslabs.github.io/partner-architecture/isv-partners/telemetry-attribution/databricks-connect | | **Databricks Connect Docs** | https://docs.databricks.com/aws/en/dev-tools/databricks-connect/python/ | | **Serverless Tutorial** | https://docs.databricks.com/aws/en/dev-tools/databricks-connect/python/tutorial-serverless | ## Requirements - **Package:** `databricks-connect` (version must match DBR on compute) - **Python:** 3.9+ - **Compute:** Serverless (recommended) or classic cluster - **Install:** `pip install -U "databricks-connect==18.0.*"` (adjust for your DBR version) - **Do NOT** install PySpark separately; `databricks-connect` includes matching version ## Authentication Decision Guide ``` Which authentication method to use? Production / automated workloads? → OAuth M2M (auth_type="oauth-m2m") ✅ RECOMMENDED User-interactive (ISV custom OAuth app)? → U2M Custom OAuth App (PKCE flow) ✅ SUPPORTED Already have an OAuth access token? → U2M Token-Env (token pass-through) ✅ SUPPORTED Local development/testing only? → PAT (token option) ⚠️ LIMITED ``` **Note:** For ISV integrations, use a custom OAuth app registered in App connections for proper branding and audit trails. ## Authentication Comparison | Method | PWAF Status | Auto Token Refresh | Browser Required | Use Case | |--------|-------------|-------------------|------------------|----------| | PAT | ⚠️ Limited | No | No | Testing only | | OAuth M2M | ✅ Recommended | Yes (SDK-native) | No | Production/automated | | U2M Custom OAuth | ✅ Supported | No | Yes | Interactive (custom app) | | U2M Token-Env | ✅ Supported | No | No | Headless/CI | ## CLIENT_ID Distinction (CRITICAL) | Variable | Purpose | Used By | |----------|---------|---------| | `DATABRICKS_CLIENT_ID` | M2M service principal UUID | OAuth M2M only | | `DATABRICKS_CLIENT_ID` | Custom OAuth app client ID | U2M custom OAuth app (different value!) | **Important:** M2M service principal client IDs and U2M custom OAuth app client IDs are different values. Use the correct one for your auth type. ## Token Lifetime Summary | Auth Type | Token TTL | Refresh Strategy | |-----------|-----------|------------------| | PAT | User-configured (90 days default) | Generate new token manually | | OAuth M2M | ~1 hour | SDK handles automatically | | U2M Custom OAuth App | ~1 hour | Re-authenticate via PKCE, recreate session | | U2M Token-Env | ~1 hour | Application must handle | --- ## Compute Options | Option | Config Parameter | Value | Notes | |--------|-----------------|-------|-------| | **Serverless (recommended)** | `serverless_compute_id` | `"auto"` | No cluster to manage | | Classic Cluster | `cluster_id` | Cluster ID | Must be running | ```python # Serverless (recommended) config = Config(host=host, token=token, serverless_compute_id="auto") # Classic cluster config = Config(host=host, token=token, cluster_id="<cluster-id>") ``` --- ## Environment Variables Reference | Variable | Required For | Description | |----------|-------------|-------------| | `DATABRICKS_HOST` | All | Workspace URL (e.g., `https://myworkspace.cloud.databricks.com`) | | `DATABRICKS_SERVERLESS_COMPUTE_ID` | Serverless | Set to `auto` for serverless compute | | `DATABRICKS_CLUSTER_ID` | Classic | Cluster ID for classic compute | | `DATABRICKS_TOKEN` | PAT | Personal access token | | `DATABRICKS_CLIENT_ID` | OAuth M2M, U2M Custom | Service principal UUID (M2M) or custom OAuth app ID (U2M) | | `DATABRICKS_CLIENT_SECRET` | OAuth M2M, U2M (if confidential) | OAuth secret | | `DATABRICKS_REDIRECT_URI` | U2M Custom OAuth (optional) | Redirect URI (default: `http://localhost:8080/callback`) | | `DATABRICKS_ACCESS_TOKEN` | U2M Token-Env | Pre-obtained OAuth access token | | `DATABRICKS_AUTH_TYPE` | Multi-auth | Auth type selector: `pat`, `oauth_m2m`, `oauth_u2m` | | `DATABRICKS_U2M_METHOD` | U2M | U2M flow: `external-browser`, `localhost`, `token-env` | --- ## User-Agent (Required for PWAF) Set via `.userAgent()` on the session builder: ```python from databricks.connect import DatabricksSession from databricks.sdk.config import Config USER_AGENT = "YourCompany_YourProduct/1.0.0" config = Config(host=host, token=token, serverless_compute_id="auto") spark = DatabricksSession.builder.sdkConfig(config).userAgent(USER_AGENT).getOrCreate() ``` **Important:** Use `.userAgent()` on the builder, NOT `product=`/`product_version=` on Config. --- ## Complete Examples ### PAT Authentication (Testing Only) ```python import os from databricks.connect import DatabricksSession from databricks.sdk.config import Config USER_AGENT = "YourCompany_YourProduct/1.0.0" def _host_url(): host = os.environ.get("DATABRICKS_HOST", "").strip() if not host: raise ValueError("Set DATABRICKS_HOST") return host if host.startswith("https://") else f"https://{host}" def get_spark_session_pat(): host = _host_url() token = os.environ.get("DATABRICKS_TOKEN", "").strip() if not token: raise ValueError("Set DATABRICKS_TOKEN") config = Config( host=host, token=token, serverless_compute_id=os.environ.get("DATABRICKS_SERVERLESS_COMPUTE_ID", "auto"), ) return DatabricksSession.builder.sdkConfig(config).userAgent(USER_AGENT).getOrCreate() spark = get_spark_session_pat() df = spark.table("samples.nyctaxi.trips") print(f"Row count: {df.count()}") ``` **Env vars:** `DATABRICKS_HOST`, `DATABRICKS_TOKEN`, `DATABRICKS_SERVERLESS_COMPUTE_ID` --- ### OAuth M2M Authentication (Production Recommended) ```python import os from databricks.connect import DatabricksSession from databricks.sdk.config import Config USER_AGENT = "YourCompany_YourProduct/1.0.0" def _host_url(): host = os.environ.get("DATABRICKS_HOST", "").strip() if not host: raise ValueError("Set DATABRICKS_HOST") return host if host.startswith("https://") else f"https://{host}" def get_spark_session_m2m(): host = _host_url() client_id = os.environ.get("DATABRICKS_CLIENT_ID", "").strip() client_secret = os.environ.get("DATABRICKS_CLIENT_SECRET", "").strip() if not client_id or not client_secret: raise ValueError("Set DATABRICKS_CLIENT_ID and DATABRICKS_CLIENT_SECRET") config = Config( host=host, client_id=client_id, client_secret=client_secret, auth_type="oauth-m2m", serverless_compute_id=os.environ.get("DATABRICKS_SERVERLESS_COMPUTE_ID", "auto"), ) return DatabricksSession.builder.sdkConfig(config).userAgent(USER_AGENT).getOrCreate() spark = get_spark_session_m2m() df = spark.table("samples.nyctaxi.trips") print(f"Row count: {df.count()}") ``` **Env vars:** `DATABRICKS_HOST`, `DATABRICKS_CLIENT_ID`, `DATABRICKS_CLIENT_SECRET`, `DATABRICKS_SERVERLESS_COMPUTE_ID` **Setup:** Create service principal in Account Console → Settings → Service principals. Generate OAuth secret. Grant access to compute and Unity Catalog. --- ### U2M Custom OAuth App (PKCE) - Recommended for ISVs ```python import os import base64 import hashlib import secrets import string import urllib.parse from http.server import HTTPServer, BaseHTTPRequestHandler from threading import Thread from urllib.parse import parse_qs, urlparse import requests from databricks.connect import DatabricksSession from databricks.sdk.config import Config USER_AGENT = "YourCompany_YourProduct/1.0.0" DEFAULT_REDIRECT_URI = "http://localhost:8080/callback" def _pkce_verifier_and_challenge(): allowed = string.ascii_letters + string.digits + "-._~" verifier = "".join(secrets.choice(allowed) for _ in range(64)) digest = hashlib.sha256(verifier.encode()).digest() challenge = base64.urlsafe_b64encode(digest).decode().rstrip("=") return verifier, challenge def run_pkce_flow(host, client_id, redirect_uri=None, client_secret=None, scope="all-apis"): redirect_uri = redirect_uri or DEFAULT_REDIRECT_URI host = host.rstrip("/") if not host.startswith("https://"): host = "https://" + host verifier, challenge = _pkce_verifier_and_challenge() state = secrets.token_urlsafe(32) auth_url = f"{host}/oidc/v1/authorize?" + urllib.parse.urlencode({ "response_type": "code", "client_id": client_id, "redirect_uri": redirect_uri, "scope": scope, "code_challenge": challenge, "code_challenge_method": "S256", "state": state, }) code_holder = [] parsed = urlparse(redirect_uri) port = parsed.port or 8080 path = parsed.path or "/callback" class Handler(BaseHTTPRequestHandler): def do_GET(self): req_path = urlparse(self.path).path if req_path == path: q = parse_qs(urlparse(self.path).query) code_holder.append(q.get("code", [None])[0]) self.send_response(200) self.send_header("Content-type", "text/html; charset=utf-8") self.end_headers() self.wfile.write(b"<html><body><h1>Authentication Successful!</h1><p>Close this tab.</p></body></html>") else: self.send_response(404) self.end_headers() def log_message(self, *args): pass server = HTTPServer(("localhost", port), Handler) thread = Thread(target=server.handle_request, daemon=True) thread.start() import webbrowser webbrowser.open(auth_url) print(f"Browser opened for sign-in. Waiting for redirect at {redirect_uri}...") thread.join(timeout=120) if not code_holder or not code_holder[0]: raise RuntimeError("No authorization code received") # Exchange code for token data = { "grant_type": "authorization_code",