Databricks Isv Python SQL Connector

Python SQL Connector Authentication (ISV)

Use this skill when implementing or testing Python SQL Connector (databricks-sql-connector) integrations for PWAF-compliant SQL query execution via a Databricks SQL warehouse.

PWAF Documentation Links

Requirements

• Package: databricks-sql-connector 2.9+
• Python: 3.8+
• SQL Warehouse: Required (DATABRICKS_HTTP_PATH)
• Install: pip install databricks-sql-connector databricks-sdk

Authentication Decision Guide

┌─────────────────────────────────────────────────────────────┐
│              Which auth method should I use?                │
└─────────────────────────────────────────────────────────────┘
                              │
                              ▼
              ┌───────────────────────────────┐
              │  Is this a production/        │
              │  automated workload?          │
              └───────────────────────────────┘
                     │              │
                    Yes            No
                     │              │
                     ▼              ▼
        ┌────────────────┐  ┌────────────────────────┐
        │  OAuth M2M     │  │  Is user interaction   │
        │  (Recommended) │  │  available?            │
        └────────────────┘  └────────────────────────┘
                                   │          │
                                  Yes        No
                                   │          │
                                   ▼          ▼
                     ┌─────────────────┐  ┌─────────────────┐
                     │  OAuth U2M      │  │  U2M Token-Env  │
                     │  (Custom OAuth) │  │  (pre-obtained) │
                     └─────────────────┘  └─────────────────┘

Authentication Comparison

ISV Note: For user-interactive flows, use U2M Custom OAuth with your registered OAuth app (Account Console → App connections). This provides custom branding, audit trails, and scoped permissions.

CLIENT_ID Distinction (CRITICAL)

Never mix these - they serve different authentication flows and will cause auth failures if confused.

Token Lifetime Summary

User-Agent (Required for ISV)

Format: <ISV-Name>_<Product-Name>/<Version>

from databricks import sql

conn = sql.connect(
    server_hostname="myworkspace.cloud.databricks.com",
    http_path="/sql/1.0/warehouses/abc123",
    access_token="...",
    user_agent_entry="YourCompany_YourProduct/1.0.0",
)

Note: Use user_agent_entry (no underscore) for connector versions 2.9+. The legacy _user_agent_entry (with underscore) is deprecated and will trigger a deprecation warning.

Environment Variables Reference

Important: Do not mix M2M and U2M environment variables. Use env -i for clean test environments.

Connection Parameters Reference

Host Normalization Helper

The connector's server_hostname expects a bare hostname (no https://):

def server_hostname(host: str) -> str:
    """Extract bare hostname from workspace URL."""
    host = host.replace("https://", "").replace("http://", "")
    if "/" in host:
        host = host.split("/")[0]
    return host

# Usage
server = server_hostname(os.environ.get("DATABRICKS_HOST", ""))

Complete Examples

PAT Authentication (Testing Only)

import os
from databricks import sql

def connect_with_pat():
    host = os.environ.get("DATABRICKS_HOST", "").strip()
    http_path = os.environ.get("DATABRICKS_HTTP_PATH", "").strip()
    token = os.environ.get("DATABRICKS_TOKEN", "").strip()
    
    server = host.replace("https://", "").split("/")[0]
    
    conn = sql.connect(
        server_hostname=server,
        http_path=http_path,
        access_token=token,
        user_agent_entry="YourCompany_YourProduct/1.0.0",
    )
    
    cursor = conn.cursor()
    cursor.execute("SELECT current_user(), current_catalog()")
    print(cursor.fetchall())
    cursor.close()
    conn.close()

Env vars: DATABRICKS_HOST, DATABRICKS_HTTP_PATH, DATABRICKS_TOKEN

OAuth M2M Authentication (Production Recommended)

Uses SDK Config with auth_type="oauth-m2m" for automatic token management:

import os
from databricks import sql
from databricks.sdk.config import Config

def connect_with_oauth_m2m():
    host = os.environ.get("DATABRICKS_HOST", "").strip()
    http_path = os.environ.get("DATABRICKS_HTTP_PATH", "").strip()
    
    host_url = host if host.startswith("https://") else f"https://{host}"
    server = host_url.replace("https://", "").split("/")[0]
    
    config = Config(
        host=host_url,
        client_id=os.environ.get("DATABRICKS_CLIENT_ID", "").strip(),
        client_secret=os.environ.get("DATABRICKS_CLIENT_SECRET", "").strip(),
        auth_type="oauth-m2m",  # Required: prevents default credential resolution
    )
    
    # CRITICAL: credentials_provider must be callable that returns callable that returns headers
    def credentials_provider():
        return lambda: config.authenticate()
    
    conn = sql.connect(
        server_hostname=server,
        http_path=http_path,
        credentials_provider=credentials_provider,
        user_agent_entry="YourCompany_YourProduct/1.0.0",
    )
    
    cursor = conn.cursor()
    cursor.execute("SELECT current_user(), current_catalog()")
    print(cursor.fetchall())
    cursor.close()
    conn.close()

Env vars: DATABRICKS_HOST, DATABRICKS_HTTP_PATH, DATABRICKS_CLIENT_ID, DATABRICKS_CLIENT_SECRET

Setup: Create service principal in Account Console → Settings → Service principals. Generate OAuth secret. Grant CAN_USE on the SQL warehouse.

OAuth U2M Custom OAuth App (Interactive)

Uses custom OAuth app with PKCE for user-interactive flows:

import os
import base64
import hashlib
import secrets
import string
import urllib.parse
import webbrowser
from http.server import HTTPServer, BaseHTTPRequestHandler
from threading import Thread
from urllib.parse import parse_qs, urlparse
import requests
from databricks import sql

DEFAULT_REDIRECT_URI = "http://localhost:8080/callback"

def pkce_verifier_and_challenge():
    """Generate PKCE code_verifier and code_challenge per RFC 7636."""
    allowed = string.ascii_letters + string.digits + "-._~"
    code_verifier = "".join(secrets.choice(allowed) for _ in range(64))
    digest = hashlib.sha256(code_verifier.encode()).digest()
    code_challenge = base64.urlsafe_b64encode(digest).decode().rstrip("=")
    return code_verifier, code_challenge

def run_u2m_flow(host, client_id, redirect_uri=None, client_secret=None, scope="all-apis"):
    """Run OAuth 2.0 authorization code flow with PKCE."""
    redirect_uri = redirect_uri or DEFAULT_REDIRECT_URI
    host = host.rstrip("/")
    if not host.startswith("https://"):
        host = "https://" + host
    
    code_verifier, code_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": code_challenge,
        "code_challenge_method": "S256",
        "state": state,
    })
    
    code_holder = []
    parsed = urlparse(redirect_uri)
    port = parsed.port or 8080
    
    class Handler(BaseHTTPRequestHandler):
        def do_GET(self):
            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")
            self.end_headers()
            self.wfile.write(b"<html><body><p>Authentication successful. Close this window.</p></body></html>")
        def log_message(self, *args): pass
    
    server = HTTPServer(("localhost", port), Handler)
    Thread(target=server.handle_request, daemon=True).start()
    webbrowser.open(auth_url)
    server.handle_request()
    
    if not code_holder[0]:
        raise RuntimeError("No authorization code received")
    
    token_url = f"{host}/oidc/v1/token"
    data = {
        "grant_type": "authorization_code",
        "code": code_holder[0],
        "redirect_uri": redirect_uri,
        "client_id": client_id,
        "code_verifier": code_verifier,
    }
    if client_secret:
        data["client_secret"] = client_secret
    
    resp = requests.post(token_url, data=data)
    resp.raise_for_status()
    return resp.json()

def connect_with_u2m():
    host = os.environ.get("DATABRICKS_HOST", "").strip()
    http_path = os.environ.get("DATABRICKS_HTTP_PATH", "").strip()
    client_id = os.environ.get("DATABRICKS_U2M_CLIENT_ID", "").strip()
    client_secret = os.environ.get("DATABRICKS_U2M_CLIENT_SECRET", "").strip() or None
    redirect_uri = os.environ.get("DATABRICKS_REDIRECT_URI", "").strip() or DEFAULT_REDIRECT_URI
    
    tokens = run_u2m_flow(host, client_id, redirect_uri, client_secret)
    access_token = tokens.get("access_token")
    
    server = host.replace("https://", "").split("/")[0]
    
    conn = sql.connect(
        server_hostname=server,
        http_path=http_path,
        access_token=access_token,
        user_agent_entry="YourCompany_YourProduct/1.0.0",
    )
    
    cursor = conn.cursor()
    cursor.execute("SELECT current_user(), current_catalog()")
    print(cursor.fetchall())
    cursor.close()
    conn.close()

Env vars: DATABRICKS_HOST, DATABRICKS_HTTP_PATH, DATABRICKS_U2M_CLIENT_ID, optional DATABRICKS_U2M_CLIENT_SECRET, DATABRICKS_REDIRECT_URI

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

OAuth U2M Token-Env (Headless/CI)

Pass a pre-obtained OAuth access token:

import os
from databricks import sql

def connect_with_token_env():
    host = os.environ.get("DATABRICKS_HOST", "").strip()
    http_path = os.environ.get("DATABRICKS_HTTP_PATH", "").strip()
    access_token = os.environ.get("DATABRICKS_ACCESS_TOKEN", "").strip()
    
    server = host.replace("https://", "").split("/")[0]
    
    conn = sql.connect(
        server_hostname=server,
        http_path=http_path,
        access_token=access_token,
        user_agent_entry="YourCompany_YourProduct/1.0.0",
    )
    
    cursor = conn.cursor()
    cursor.execute("SELECT current_user(), current_catalog()")
    print(cursor.fetchall())
    cursor.close()
    conn.close()

Env vars: DATABRICKS_HOST, DATABRICKS_HTTP_PATH, DATABRICKS_ACCESS_TOKEN

Error Handling Patterns

Error Classification

def classify_error(error: Exception) -> tuple[str, bool]:
    """Classify error into category and retryable flag."""
    error_str = str(error).lower()
    
    # Authentication errors (non-retryable)
    if "401" in error_str or "unauthorized" in error_str:
        return "authentication", False
    if "invalid_client" in error_str or "invalid_grant" in error_str:
        return "authentication", False
    
    # Permission errors (non-retryable)
    if "403" in error_str or "permission_denied" in error_str:
        return "permission", False
    
    # SQL syntax errors (non-retryable)
    if "analysisexception" in error_str or "parseexception" in error_str:
        return "syntax", False
    
    # Resource not found (non-retryable)
    if "table_or_view_not_found" in error_str or "catalog_not_found" in error_str:
        return "not_found", False
    
    # Rate limiting (retryable)
    if "429" in error_str or "rate limit" in error_str:
        return "rate_limit", True
    
    # Warehouse unavailable (retryable)
    if "503" in error_str or "temporarily_unavailable" in error_str:
        return "warehouse", True
    
    # Network errors (retryable)
    if "connection" in error_str and ("reset" in error_str or "refused" in error_str):
        return "network", True
    
    return "unknown", False

Retry with Exponential Backoff

import time
import random

def retry_query(cursor, query: str, max_retries: int = 3, initial_backoff: float = 2.0):
    """Execute query with exponential backoff retry."""
    last_error = None
    backoff = initial_backoff
    
    for attempt in range(max_retries + 1):
        try:
            cursor.execute(query)
            return cursor.fetchall()
        except Exception as e:
            last_error = e
            category, retryable = classify_error(e)
            
            if not retryable or attempt >= max_retries:
                raise
            
            jitter = backoff * 0.25 * random.random()
            sleep_time = backoff + jitter
            print(f"[Retry {attempt + 1}/{max_retries}] Waiting {sleep_time:.1f}s...")
            time.sleep(sleep_time)
            backoff = min(backoff * 2, 60.0)
    
    raise last_error

Multi-Auth Pattern

Support multiple auth types in a single script using APP_AUTH_TYPE:

import os
from databricks import sql
from databricks.sdk.config import Config

def get_connection():
    auth_type = os.environ.get("APP_AUTH_TYPE", "oauth_m2m").lower()
    host = os.environ.get("DATABRICKS_HOST", "").strip()
    http_path = os.environ.get("DATABRICKS_HTTP_PATH", "").strip()
    server = host.replace("https://", "").split("/")[0]
    
    if auth_type == "pat":
        return sql.connect(
            server_hostname=server,
            http_path=http_path,
            access_token=os.environ.get("DATABRICKS_TOKEN", ""),
            user_agent_entry="YourCompany_YourProduct/1.0.0",
        )
    
    elif auth_type in ("oauth_m2m", "m2m"):
        host_url = host if host.startswith("https://") else f"https://{host}"
        config = Config(
            host=host_url,
            client_id=os.environ.get("DATABRICKS_CLIENT_ID", ""),
            client_secret=os.environ.get("DATABRICKS_CLIENT_SECRET", ""),
            auth_type="oauth-m2m",
        )
        def credentials_provider():
            return lambda: config.authenticate()
        return sql.connect(
            server_hostname=server,
            http_path=http_path,
            credentials_provider=credentials_provider,
            user_agent_entry="YourCompany_YourProduct/1.0.0",
        )
    
    elif auth_type in ("oauth_u2m", "u2m"):
        tokens = run_u2m_flow(
            host=host,
            client_id=os.environ.get("DATABRICKS_U2M_CLIENT_ID", ""),
            client_secret=os.environ.get("DATABRICKS_U2M_CLIENT_SECRET", "") or None,
        )
        return sql.connect(
            server_hostname=server,
            http_path=http_path,
            access_token=tokens["access_token"],
            user_agent_entry="YourCompany_YourProduct/1.0.0",
        )
    
    raise ValueError(f"Unknown auth type: {auth_type}")

Note: Use APP_AUTH_TYPE (not DATABRICKS_AUTH_TYPE) to avoid conflicts with SDK's internal auth_type parameter.

Auth Isolation

Run tests with a clean environment using env -i:

# PAT test
env -i PATH=$PATH HOME=$HOME \
  DATABRICKS_HOST=$DATABRICKS_HOST \
  DATABRICKS_HTTP_PATH=$DATABRICKS_HTTP_PATH \
  DATABRICKS_TOKEN=$DATABRICKS_TOKEN \
  python your_script.py

# OAuth M2M test
env -i PATH=$PATH HOME=$HOME \
  DATABRICKS_HOST=$DATABRICKS_HOST \
  DATABRICKS_HTTP_PATH=$DATABRICKS_HTTP_PATH \
  DATABRICKS_CLIENT_ID=$DATABRICKS_CLIENT_ID \
  DATABRICKS_CLIENT_SECRET=$DATABRICKS_CLIENT_SECRET \
  python your_script.py

# OAuth U2M test (needs DISPLAY, USER for browser)
env -i PATH=$PATH HOME=$HOME USER=$USER DISPLAY=$DISPLAY \
  DATABRICKS_HOST=$DATABRICKS_HOST \
  DATABRICKS_HTTP_PATH=$DATABRICKS_HTTP_PATH \
  DATABRICKS_U2M_CLIENT_ID=$DATABRICKS_U2M_CLIENT_ID \
  python your_script.py

Redirect URI Reference

Validation Query

Verify User-Agent telemetry is being recorded correctly:

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;

Troubleshooting

credentials_provider Errors

Authentication Errors

Permission Errors

Warehouse Errors

U2M Browser Issues

Implementation Learnings

1. credentials_provider signature: Must be Callable[[], Callable[[], Dict[str, str]]] - a callable that returns a callable that returns headers. Use lambda: config.authenticate pattern.

2. auth_type for M2M: Always pass auth_type="oauth-m2m" to SDK Config when using client_id/client_secret to prevent default credential resolution.

3. CLIENT_ID confusion: Use separate env vars (DATABRICKS_CLIENT_ID for M2M, DATABRICKS_U2M_CLIENT_ID for U2M) to avoid mixing up service principal UUIDs with OAuth app client IDs.

4. APP_AUTH_TYPE vs DATABRICKS_AUTH_TYPE: Use APP_AUTH_TYPE for your app's auth selector to avoid conflicts with SDK's internal auth_type parameter.

5. Host normalization: SQL connector expects bare hostname without https://. Always strip scheme before passing to server_hostname.

6. User-Agent parameter: Use user_agent_entry (no underscore) for connector versions 2.9+. The legacy _user_agent_entry is deprecated.

PKCE Flow Implementation Details

When implementing custom OAuth U2M PKCE flows, follow these patterns for reliable browser callback handling:

Use threading.Event for callback signaling:

from threading import Thread, Event

callback_received = Event()
CALLBACK_TIMEOUT = 120  # seconds

def serve_until_callback():
    while not callback_received.is_set():
        server.handle_request()

thread = Thread(target=serve_until_callback, daemon=True)
thread.start()

Handle multiple HTTP requests (browsers send favicon, prefetch, etc.):

class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        req_path = urlparse(self.path).path
        if req_path == "/callback":  # Only process OAuth callback
            code = parse_qs(urlparse(self.path).query).get("code", [None])[0]
            if code:
                callback_received.set()
                self.send_success_response()
        else:
            self.send_response(404)
            self.end_headers()

Use output flushing for immediate feedback:

print("Waiting for browser authentication...", flush=True)

Wait with timeout:

if not callback_received.wait(timeout=CALLBACK_TIMEOUT):
    raise RuntimeError(f"Authentication timed out after {CALLBACK_TIMEOUT}s")

Key insight: Using server.handle_request() only once can be consumed by browser prefetch requests (like favicon), causing the actual OAuth callback to be missed. The Event-based loop pattern ensures all requests are handled until the actual callback arrives.

Key Differences from Databricks SDK for Python