databricks-isv-python-dbconnect

name	databricks-isv-python-dbconnect
description	PWAF-compliant Databricks Connect for Python: PAT, OAuth M2M, OAuth U2M (PKCE + token-env), userAgent telemetry. Use when building Spark applications that run on Databricks serverless or classic compute.

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

# 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:

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)

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)

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

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",