Databricks Isv Python Dbconnect

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

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

CLIENT_ID Distinction (CRITICAL)

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

Compute Options

# 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

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",
        "code": code_holder[0],
        "redirect_uri": redirect_uri,
        "client_id": client_id,
        "code_verifier": verifier,
    }
    if client_secret:
        data["client_secret"] = client_secret
    
    resp = requests.post(f"{host}/oidc/v1/token", data=data,
                        headers={"Content-Type": "application/x-www-form-urlencoded"})
    resp.raise_for_status()
    return resp.json()

def get_spark_session_u2m():
    host = os.environ.get("DATABRICKS_HOST", "").strip()
    client_id = os.environ.get("DATABRICKS_CLIENT_ID", "").strip()
    redirect_uri = os.environ.get("DATABRICKS_REDIRECT_URI", "").strip() or DEFAULT_REDIRECT_URI
    client_secret = os.environ.get("DATABRICKS_CLIENT_SECRET", "").strip() or None
    
    if not host or not client_id:
        raise ValueError("Set DATABRICKS_HOST and DATABRICKS_CLIENT_ID")
    
    tokens = run_pkce_flow(host, client_id, redirect_uri, client_secret)
    access_token = tokens.get("access_token")
    if not access_token:
        raise ValueError("No access_token in response")
    
    host_url = host if host.startswith("https://") else f"https://{host}"
    config = Config(
        host=host_url,
        token=access_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_u2m()
df = spark.table("samples.nyctaxi.trips")
print(f"Row count: {df.count()}")

Env vars: DATABRICKS_HOST, DATABRICKS_CLIENT_ID, DATABRICKS_SERVERLESS_COMPUTE_ID

Dependencies: pip install requests

Setup: Create OAuth app in Account Console → Settings → App connections. Add redirect URI http://localhost:8080/callback.

U2M Token-Env Authentication (Headless/CI)

import os
from databricks.connect import DatabricksSession
from databricks.sdk.config import Config

USER_AGENT = "YourCompany_YourProduct/1.0.0"

def get_spark_session_token_env():
    host = os.environ.get("DATABRICKS_HOST", "").strip()
    token = os.environ.get("DATABRICKS_ACCESS_TOKEN", "").strip() or \
            os.environ.get("DATABRICKS_TOKEN", "").strip()
    
    if not host:
        raise ValueError("Set DATABRICKS_HOST")
    if not token:
        raise ValueError("Set DATABRICKS_ACCESS_TOKEN or DATABRICKS_TOKEN")
    
    host_url = host if host.startswith("https://") else f"https://{host}"
    config = Config(
        host=host_url,
        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_token_env()
df = spark.table("samples.nyctaxi.trips")
print(f"Row count: {df.count()}")

Env vars: DATABRICKS_HOST, DATABRICKS_ACCESS_TOKEN, DATABRICKS_SERVERLESS_COMPUTE_ID

Error Handling Patterns

Error Classification

def classify_error(error):
    msg = str(error).lower()
    
    if "401" in msg or "unauthorized" in msg or "invalid_client" in msg:
        return {"type": "AUTHENTICATION_ERROR", "retryable": False,
                "action": "Check DATABRICKS_TOKEN or OAuth credentials"}
    if "403" in msg or "permission denied" in msg or "forbidden" in msg:
        return {"type": "PERMISSION_DENIED", "retryable": False,
                "action": "Grant service principal access to compute and Unity Catalog"}
    if "table" in msg and "not found" in msg:
        return {"type": "RESOURCE_NOT_FOUND", "retryable": False,
                "action": "Check table name exists and is accessible"}
    if "syntax error" in msg or "analysisexception" in msg:
        return {"type": "SQL_SYNTAX_ERROR", "retryable": False,
                "action": "Fix SQL syntax"}
    if "429" in msg or "rate limit" in msg or "too many requests" in msg:
        return {"type": "RATE_LIMITED", "retryable": True,
                "action": "Implement exponential backoff"}
    if "503" in msg or "service unavailable" in msg:
        return {"type": "SERVICE_UNAVAILABLE", "retryable": True,
                "action": "Wait and retry with backoff"}
    if "connection" in msg or "network" in msg or "timeout" in msg:
        return {"type": "NETWORK_ERROR", "retryable": True,
                "action": "Check network connectivity"}
    return {"type": "UNKNOWN_ERROR", "retryable": False, "action": "Check full error message"}

def is_retryable(error):
    return classify_error(error).get("retryable", False)

Retry Pattern with Exponential Backoff

import time

def execute_with_retry(func, max_retries=3, initial_backoff=1.0, max_backoff=30.0):
    backoff = initial_backoff
    last_error = None
    
    for attempt in range(max_retries + 1):
        try:
            return func()
        except Exception as e:
            last_error = e
            if not is_retryable(e) or attempt >= max_retries:
                raise
            print(f"[Retry {attempt + 1}/{max_retries}] Waiting {backoff}s. Error: {e}")
            time.sleep(backoff)
            backoff = min(backoff * 2, max_backoff)
    
    raise last_error

Multi-Auth Pattern

Support multiple auth types using DATABRICKS_AUTH_TYPE:

import os
from databricks.connect import DatabricksSession
from databricks.sdk.config import Config

USER_AGENT = "YourCompany_YourProduct/1.0.0"

def resolve_auth_type():
    v = (os.environ.get("DATABRICKS_AUTH_TYPE") or "oauth_m2m").strip().lower()
    aliases = {"1": "pat", "2": "oauth_m2m", "m2m": "oauth_m2m",
               "3": "oauth_u2m", "u2m": "oauth_u2m"}
    return aliases.get(v, v)

def get_spark_session():
    auth_type = resolve_auth_type()
    host = os.environ.get("DATABRICKS_HOST", "").strip()
    if not host.startswith("https://"):
        host = f"https://{host}"
    
    compute_kw = {"serverless_compute_id": os.environ.get("DATABRICKS_SERVERLESS_COMPUTE_ID", "auto")}
    
    if auth_type == "pat":
        token = os.environ.get("DATABRICKS_TOKEN", "").strip()
        config = Config(host=host, token=token, **compute_kw)
    
    elif auth_type == "oauth_m2m":
        config = Config(
            host=host,
            client_id=os.environ.get("DATABRICKS_CLIENT_ID", "").strip(),
            client_secret=os.environ.get("DATABRICKS_CLIENT_SECRET", "").strip(),
            auth_type="oauth-m2m",
            **compute_kw
        )
    
    elif auth_type == "oauth_u2m":
        token = os.environ.get("DATABRICKS_ACCESS_TOKEN", "").strip() or \
                os.environ.get("DATABRICKS_TOKEN", "").strip()
        if not token:
            # Run PKCE flow to get token
            token = run_pkce_flow(host, os.environ.get("DATABRICKS_CLIENT_ID", "")).get("access_token")
        config = Config(host=host, token=token, **compute_kw)
    
    else:
        raise ValueError(f"Unsupported DATABRICKS_AUTH_TYPE: {auth_type}")
    
    return DatabricksSession.builder.sdkConfig(config).userAgent(USER_AGENT).getOrCreate()

U2M Token Refresh Pattern

For long-running U2M applications, track token expiry and refresh proactively:

import os
import time
from databricks.connect import DatabricksSession
from databricks.sdk.config import Config

USER_AGENT = "YourCompany_YourProduct/1.0.0"
REFRESH_BUFFER_SEC = 5 * 60  # Refresh 5 min before expiry

class U2MSessionManager:
    def __init__(self):
        self.spark = None
        self.token_expires_at = 0
    
    def _obtain_token(self):
        tokens = run_pkce_flow(
            host=os.environ.get("DATABRICKS_HOST", "").strip(),
            client_id=os.environ.get("DATABRICKS_CLIENT_ID", "").strip(),
            redirect_uri=os.environ.get("DATABRICKS_REDIRECT_URI", ""),
            client_secret=os.environ.get("DATABRICKS_CLIENT_SECRET", ""),
        )
        self.token_expires_at = time.time() + 55 * 60  # ~55 min TTL
        return tokens.get("access_token")
    
    def _create_session(self, token):
        host = os.environ.get("DATABRICKS_HOST", "").strip()
        if not host.startswith("https://"):
            host = f"https://{host}"
        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()
    
    def get_session(self):
        if time.time() >= self.token_expires_at - REFRESH_BUFFER_SEC:
            if self.spark:
                self.spark.stop()
            token = self._obtain_token()
            self.spark = self._create_session(token)
        return self.spark
    
    def shutdown(self):
        if self.spark:
            self.spark.stop()
            self.spark = None

Config Parameters Reference

DatabricksSession.builder Options

databricks.sdk.config.Config Parameters

Config Examples by Auth Type

from databricks.sdk.config import Config

# PAT Authentication
config_pat = Config(
    host="https://myworkspace.cloud.databricks.com",
    token="dapi...",
    serverless_compute_id="auto",
)

# OAuth M2M Authentication
config_m2m = Config(
    host="https://myworkspace.cloud.databricks.com",
    client_id="<service-principal-uuid>",
    client_secret="<oauth-secret>",
    auth_type="oauth-m2m",
    serverless_compute_id="auto",
)

# U2M Token Pass-through
config_u2m = Config(
    host="https://myworkspace.cloud.databricks.com",
    token="<pre-obtained-oauth-token>",
    serverless_compute_id="auto",
)

# Classic Cluster (instead of serverless)
config_classic = Config(
    host="https://myworkspace.cloud.databricks.com",
    token="dapi...",
    cluster_id="0123-456789-abc123",
)

Host Normalization

The SDK Config expects the full URL with https://:

def normalize_host(host):
    host = host.strip().rstrip("/")
    if not host.startswith("https://"):
        host = "https://" + host
    return host

Redirect URI Reference

Registering Redirect URIs

1. Go to Account Console → Settings → App connections
2. Select your OAuth app or create new
3. Add redirect URI: http://localhost:8080/callback
4. For production deployments, add your hosted callback URL

Custom Redirect URI Example

import os

# Default for local development
DEFAULT_REDIRECT_URI = "http://localhost:8080/callback"

# Allow override via environment
redirect_uri = os.environ.get("DATABRICKS_REDIRECT_URI", "").strip() or DEFAULT_REDIRECT_URI

# Use in PKCE flow
tokens = run_pkce_flow(host, client_id, redirect_uri=redirect_uri)

Auth Isolation

Run tests with a clean environment using env -i to avoid credential conflicts between auth types.

PAT Test

env -i PATH=$PATH HOME=$HOME \
  DATABRICKS_HOST=$DATABRICKS_HOST \
  DATABRICKS_TOKEN=$DATABRICKS_TOKEN \
  DATABRICKS_SERVERLESS_COMPUTE_ID=auto \
  python your_script.py

OAuth M2M Test

env -i PATH=$PATH HOME=$HOME \
  DATABRICKS_HOST=$DATABRICKS_HOST \
  DATABRICKS_CLIENT_ID=$DATABRICKS_CLIENT_ID \
  DATABRICKS_CLIENT_SECRET=$DATABRICKS_CLIENT_SECRET \
  DATABRICKS_SERVERLESS_COMPUTE_ID=auto \
  python your_script.py

U2M Custom OAuth App Test

env -i PATH=$PATH HOME=$HOME \
  DATABRICKS_HOST=$DATABRICKS_HOST \
  DATABRICKS_CLIENT_ID=$DATABRICKS_U2M_CLIENT_ID \
  DATABRICKS_CLIENT_SECRET=$DATABRICKS_U2M_CLIENT_SECRET \
  DATABRICKS_REDIRECT_URI="http://localhost:8080/callback" \
  DATABRICKS_SERVERLESS_COMPUTE_ID=auto \
  python your_script.py

U2M Token-Env Test

env -i PATH=$PATH HOME=$HOME \
  DATABRICKS_HOST=$DATABRICKS_HOST \
  DATABRICKS_ACCESS_TOKEN=$MY_OAUTH_TOKEN \
  DATABRICKS_SERVERLESS_COMPUTE_ID=auto \
  python your_script.py

Why Auth Isolation?

Mixing credentials can cause unexpected behavior:

Troubleshooting

Common Errors

OAuth-Specific Errors

Validation Query

Verify User-Agent telemetry:

SELECT event_time, user_agent, action_name, request_params
FROM system.access.audit
WHERE event_time > current_timestamp() - INTERVAL 1 HOUR
  AND lower(user_agent) LIKE '%yourcompany%'
ORDER BY event_time DESC
LIMIT 10;

Key Differences from Other Connectors