# Databricks Isv Python SQL Connector

> PWAF-compliant Python SQL Connector (databricks-sql-connector): PAT, OAuth M2M, OAuth U2M (custom OAuth app PKCE + token-env), credentials_provider patterns, error handling, retry logic. Use when building Python integrations that run SQL queries via a Databricks SQL warehouse.

- Skill: `databricks-solutions/databricks-isv-python-sql-connector` (Agent Skill, multi-file: 2 files)
- Install (CLI): `npx skillmds@latest add databricks-solutions/databricks-isv-python-sql-connector`
- Raw SKILL.md: https://api.skillmd.com/api/skills/databricks-solutions/databricks-isv-python-sql-connector/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Data & Analytics
- Author: databricks-solutions (https://skillmd.com/u/databricks-solutions)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/databricks-solutions/databricks-isv-python-sql-connector

---


<!-- skill-version: 1.0.0 -->

# 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

| 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 |
| **SQL Connector Telemetry** | https://databrickslabs.github.io/partner-architecture/isv-partners/telemetry-attribution/sql-drivers |
| **Databricks SQL Connector for Python** | https://docs.databricks.com/aws/en/dev-tools/python-sql-connector |
| **PyPI** | https://pypi.org/project/databricks-sql-connector/ |

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

| Method | PWAF Status | Auto Token Refresh | Browser Required | Use Case |
|--------|-------------|-------------------|------------------|----------|
| PAT | ⚠️ Limited | No | No | Testing only |
| OAuth M2M | ✅ Recommended | Yes (via SDK Config) | No | Production/automated |
| U2M Custom OAuth | ✅ Recommended (ISV) | No | Yes | Interactive (custom app) |
| U2M Token-Env | ✅ Supported | No | No | Headless/CI |

**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)

| Variable | Purpose | Where to Find |
|----------|---------|---------------|
| `DATABRICKS_CLIENT_ID` | Service Principal UUID for **M2M** | Workspace → Settings → Identity and access → Service principals |
| `DATABRICKS_U2M_CLIENT_ID` | Custom OAuth App client ID for **U2M** | Account Console → Settings → App connections |

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

---

## Token Lifetime Summary

| Auth Type | Access Token | Refresh Token | Auto-Refresh |
|-----------|-------------|---------------|--------------|
| PAT | Configurable (default 90 days) | N/A | No |
| OAuth M2M | 1 hour | N/A (uses client credentials) | Yes (SDK Config) |
| OAuth U2M | 1 hour | Up to 90 days | No (manual) |

---

## User-Agent (Required for ISV)

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

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

| Variable | Required For | Description |
|----------|-------------|-------------|
| `DATABRICKS_HOST` | All | Workspace URL (e.g., `https://myworkspace.cloud.databricks.com`) |
| `DATABRICKS_HTTP_PATH` | All | SQL warehouse HTTP path (e.g., `/sql/1.0/warehouses/abc123`) |
| `DATABRICKS_TOKEN` | PAT | Personal access token |
| `DATABRICKS_CLIENT_ID` | OAuth M2M | Service principal UUID |
| `DATABRICKS_CLIENT_SECRET` | OAuth M2M | Service principal OAuth secret |
| `DATABRICKS_U2M_CLIENT_ID` | U2M Custom OAuth | Custom OAuth app client ID from App connections |
| `DATABRICKS_U2M_CLIENT_SECRET` | U2M Custom OAuth (optional) | Custom OAuth app client secret |
| `DATABRICKS_REDIRECT_URI` | U2M Custom OAuth (optional) | Custom redirect URI (default: `http://localhost:8080/callback`) |
| `DATABRICKS_ACCESS_TOKEN` | U2M Token-Env | Pre-obtained OAuth access token |
| `APP_AUTH_TYPE` | Multi-auth | Auth type selector: `pat`, `oauth_m2m`, `oauth_u2m` |

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

---

## Connection Parameters Reference

| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| `server_hostname` | str | Yes | Workspace hostname (no `https://`) |
| `http_path` | str | Yes | SQL warehouse HTTP path |
| `access_token` | str | PAT, U2M | Token for authentication |
| `credentials_provider` | callable | OAuth M2M | Callable that returns callable that returns headers |
| `user_agent_entry` | str | Recommended | ISV partner telemetry string |

---

## Host Normalization Helper

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

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

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

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

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

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

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

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

```python
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}")
```

| `APP_AUTH_TYPE` | Auth flow | Env vars required |
|-----------------|-----------|-------------------|
| `pat` | Personal Access Token | `DATABRICKS_TOKEN` |
| `oauth_m2m` or `m2m` | OAuth M2M (client credentials) | `DATABRICKS_CLIENT_ID`, `DATABRICKS_CLIENT_SECRET` |
| `oauth_u2m` or `u2m` | OAuth U2M (custom OAuth app) | `DATABRICKS_U2M_CLIENT_ID`, optional `DATABRICKS_U2M_CLIENT_SECRET` |

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

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

| Environment | Recommended URI | Notes |
|-------------|-----------------|-------|
| Local development | `http://localhost:8080/callback` | Register in App connections |
| Docker/container | `http://host.docker.internal:8080/callback` | For Docker Desktop |
| Production web app | `https://your-app.com/oauth/callback` | HTTPS required |

---

## Validation Query

Verify User-Agent telemetry is being recorded correctly:

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

---

## Troubleshooting

### credentials_provider Errors

| Error | Cause | Solution |
|-------|-------|----------|
| `'dict' object is not callable` | credentials_provider passed incorrectly | Use `lambda: config.authenticate` (callable that returns callable) |
| `Cannot configure default credentials` | M2M Config without auth_type | Pass `auth_type="oauth-m2m"` to Config |
| `More than one authorization method` | PAT and M2M env vars both set | Run with clean env (`env -i`) for one auth type only |

### Authentication Errors

| Error | Cause | Solution |
|-------|-------|----------|
| `401 Unauthorized` | Invalid credentials | Check token/secret validity and expiration |
| `invalid_client` | Wrong client ID | Verify CLIENT_ID matches registered app |
| `invalid_grant` | Auth code expired/reused | Auth codes are single-use; restart flow |

### Permission Errors

| Error | Cause | Solution |
|-------|-------|----------|
| `403 PERMISSION_DENIED` | Missing access | Grant CAN_USE on warehouse, USE_CATALOG/SCHEMA |
| `TABLE_OR_VIEW_NOT_FOUND` | Missing table permission | Grant SELECT on table |

### Warehouse Errors

| Error | Cause | Solution |
|-------|-------|----------|
| `503 Service Unavailable` | Warehouse starting | Wait and retry with backoff |
| `TEMPORARILY_UNAVAILABLE` | Warehouse busy | Retry with exponential backoff |

### U2M Browser Issues

| Symptom | Cause | Solution |
|---------|-------|----------|
| Browser doesn't open | Missing env vars | Pass `HOME`, `USER`, `DISPLAY` via `env -i` |
| Port in use | Another process | Use different port in redirect URI |
| Redirect error | URI mismatch | Verify redirect URI matches exactly in App connections |

---

## 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:**
```python
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.):**
```python
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:**
```python
print("Waiting for browser authentication...", flush=True)
```

**Wait with timeout:**
```python
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

| Aspect | Python SQL Connector | Databricks SDK for Python |
|--------|---------------------|--------------------------|
| **Purpose** | SQL queries via warehouse | REST APIs (UC, Jobs, Clusters) |
| **SQL Warehouse** | Required (`http_path`) | Not needed |
| **Interface** | DBAPI 2.0 cursor | Typed Python methods |
| **Auth Config** | `sql.connect()` parameters | `Config()` object |
| **User-Agent** | `user_agent_entry` (per-connection, v2.9+) | `useragent.with_partner()` (global) |
| **M2M Token Refresh** | Via SDK Config's credentials_provider | Automatic |
| **U2M Browser** | Not built-in (use PKCE helper) | Config with `auth_type="external-browser"` |