# Databricks Isv Python Dbconnect

> 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.

- Skill: `databricks-solutions/databricks-isv-python-dbconnect` (Agent Skill, multi-file: 2 files)
- Install (CLI): `npx skillmds@latest add databricks-solutions/databricks-isv-python-dbconnect`
- Raw SKILL.md: https://api.skillmd.com/api/skills/databricks-solutions/databricks-isv-python-dbconnect/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: DevOps & Infra
- Author: databricks-solutions (https://skillmd.com/u/databricks-solutions)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/databricks-solutions/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",
        "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)

```python
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

```python
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

```python
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`:

```python
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()
```

| `DATABRICKS_AUTH_TYPE` | Auth flow | Env vars required |
|------------------------|-----------|-------------------|
| `pat` (or `1`) | Personal Access Token | `DATABRICKS_TOKEN` |
| `oauth_m2m` (or `m2m`, `2`) | OAuth M2M (client credentials) | `DATABRICKS_CLIENT_ID`, `DATABRICKS_CLIENT_SECRET` |
| `oauth_u2m` (or `u2m`, `3`) | OAuth U2M (PKCE or token-env) | `DATABRICKS_CLIENT_ID` or `DATABRICKS_ACCESS_TOKEN` |

---

## U2M Token Refresh Pattern

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

```python
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

| Method | Description | Example |
|--------|-------------|---------|
| `.sdkConfig(config)` | Pass SDK Config object | `.sdkConfig(config)` |
| `.userAgent(entry)` | **Required for PWAF** | `.userAgent("YourCompany_Product/1.0.0")` |
| `.getOrCreate()` | Create or get existing session | `.getOrCreate()` |

### databricks.sdk.config.Config Parameters

| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| `host` | str | Yes | Workspace URL (include `https://`) |
| `token` | str | PAT, U2M | Personal access token or OAuth access token |
| `client_id` | str | M2M | Service principal UUID |
| `client_secret` | str | M2M | Service principal OAuth secret |
| `auth_type` | str | M2M | Set to `"oauth-m2m"` for client credentials flow |
| `serverless_compute_id` | str | Serverless | Set to `"auto"` for serverless compute |
| `cluster_id` | str | Classic | Cluster ID for classic compute |

### Config Examples by Auth Type

```python
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://`:

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

---

## Redirect URI Reference

| Scenario | Default Redirect URI | Notes |
|----------|---------------------|-------|
| Custom OAuth app (Python default) | `http://localhost:8080/callback` | Default in PKCE helper |
| Custom OAuth app (configured) | Via `DATABRICKS_REDIRECT_URI` | Must match App connections registration |
| SDK external-browser | Dynamic localhost port | SDK picks available port automatically |

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

```python
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

```bash
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

```bash
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

```bash
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

```bash
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:

| Problem | Cause | Solution |
|---------|-------|----------|
| M2M uses wrong identity | `DATABRICKS_TOKEN` also set | Use `env -i` to clear PAT |
| U2M browser flow fails | `DATABRICKS_CLIENT_ID` set to M2M SP | Use U2M-specific client ID |
| Wrong auth type selected | Multiple credential sets present | Set only required vars |

---

## Troubleshooting

### Common Errors

| Error | Cause | Solution |
|-------|-------|----------|
| `No compute found` | Missing compute config | Set `DATABRICKS_SERVERLESS_COMPUTE_ID=auto` or `DATABRICKS_CLUSTER_ID` |
| `401 Unauthorized` | Invalid credentials | Check token or OAuth secrets |
| `403 Forbidden` | Missing permissions | Grant SP access to compute and Unity Catalog |
| `Version mismatch` | databricks-connect version doesn't match DBR | Install matching version |
| `external-browser` auth fails | M2M client_id set | Unset `DATABRICKS_CLIENT_ID` for built-in external-browser |
| `Connection refused` | Cluster not running | Start cluster or use serverless |

### OAuth-Specific Errors

| Error | Cause | Solution |
|-------|-------|----------|
| `OAuth application not available` | Account not enabled for OAuth | Enable OAuth in account console |
| `redirect_uri mismatch` | URI not registered | Add redirect URI in App connections |
| `invalid_client` | Wrong client ID/secret | Verify credentials match registered OAuth app |
| `invalid_grant` | Auth code expired/reused | Auth codes are single-use; restart OAuth flow |

---

## Validation Query

Verify User-Agent telemetry:

```sql
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

| Aspect | Databricks Connect | Python SQL Connector | Java JDBC |
|--------|-------------------|---------------------|-----------|
| **Package** | `databricks-connect` | `databricks-sql-connector` | `databricks-jdbc` |
| **Use Case** | Spark DataFrames | SQL queries | SQL queries |
| **Compute** | Serverless/Cluster | SQL Warehouse | SQL Warehouse |
| **User-Agent** | `.userAgent()` on builder | `_user_agent_entry` param | `UserAgentEntry` URL param |
| **M2M Config** | `auth_type="oauth-m2m"` | `access_token` (manual) | `AuthMech=11`, `Auth_Flow=1` |
| **Session** | `DatabricksSession` | `Connection` | JDBC `Connection` |