databricks-isv-python-sql-connector

name	databricks-isv-python-sql-connector
description	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.

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>

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://):

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)