Databricks SDK for Python cheat sheet for Azure admins - auth, clusters, jobs, permissions, SQL and errors

The SDK wraps the REST API with typed methods, automatic pagination and retries. Install it with pip install databricks-sdk. It's also preinstalled on recent Databricks runtimes, but often at an older version. Method and field names occasionally change between releases, so pin the version in automation and check the SDK reference when a call fails with TypeError or AttributeError.

#Authentication

from databricks.sdk import WorkspaceClient, AccountClient

w = WorkspaceClient(profile="prod")
print(w.config.auth_type, w.current_user.me().user_name)

# Account-level APIs on Azure
a = AccountClient(host="https://accounts.azuredatabricks.net",
                  account_id="<account-id>", auth_type="azure-cli")
for ws in a.workspaces.list():
    print(ws.workspace_id, ws.workspace_name)

Tip

Print w.config.auth_type at the start of every script. When several credentials are present, the SDK picks one by its own resolution order. Printing it shows you whether you are running as yourself or as the service principal.

#Clusters

from datetime import timedelta

for c in w.clusters.list():
    print(c.cluster_id, c.cluster_name, c.state, c.creator_user_name)

w.clusters.ensure_cluster_is_running("0123-456789-abcde123")   # starts it if needed and waits
w.clusters.start(cluster_id="0123-456789-abcde123").result(timeout=timedelta(minutes=20))
w.clusters.delete(cluster_id="0123-456789-abcde123")            # terminates; it does not remove the cluster
w.clusters.permanent_delete(cluster_id="0123-456789-abcde123")  # removes it for good

Find running all-purpose clusters without auto-termination:

from databricks.sdk.service.compute import State
for c in w.clusters.list():
    if c.state == State.RUNNING and not c.autotermination_minutes:
        print("no auto-termination:", c.cluster_name, c.creator_user_name)

Job clusters also show up in clusters.list(). Filter on c.cluster_source if you only want interactive ones.

#Jobs and runs

for j in w.jobs.list(name="gold_daily"):          # paginates automatically
    print(j.job_id, j.settings.name)

run = w.jobs.run_now(job_id=123, job_parameters={"date": "2026-11-09"})
result = run.result(timeout=timedelta(hours=2))   # waits for the run to finish
print(result.state.result_state, result.run_page_url)

for r in w.jobs.list_runs(job_id=123, active_only=True):
    print(r.run_id, r.state.life_cycle_state)

w.jobs.cancel_run(run_id=987654)
w.jobs.repair_run(run_id=987654, rerun_all_failed_tasks=True)

Pause every schedule in a workspace, for example before maintenance:

from databricks.sdk.service.jobs import JobSettings, PauseStatus

paused = []
for j in w.jobs.list():
    s = w.jobs.get(job_id=j.job_id).settings
    if s.schedule and s.schedule.pause_status != PauseStatus.PAUSED:
        s.schedule.pause_status = PauseStatus.PAUSED
        w.jobs.update(job_id=j.job_id, new_settings=JobSettings(schedule=s.schedule))
        paused.append(j.job_id)
print(paused)    # save this list so you unpause only these jobs afterwards

Warning

Jobs that a bundle or Terraform deploys get their schedules back on the next deploy. Pause deployments as well, or set the pause status in code.

#Permissions (workspace objects)

acl = w.permissions.get(request_object_type="jobs", request_object_id="123")
for e in acl.access_control_list:
    print(e.group_name or e.user_name or e.service_principal_name,
          [p.permission_level for p in e.all_permissions])

from databricks.sdk.service.iam import AccessControlRequest, PermissionLevel
w.permissions.update(
    request_object_type="jobs", request_object_id="123",
    access_control_list=[AccessControlRequest(group_name="data-platform-ops",
                                              permission_level=PermissionLevel.CAN_MANAGE)])

update adds to the existing ACL. set replaces the whole list, so it can remove your own access. Other object types include clusters, cluster-policies, instance-pools, notebooks, directories, pipelines, sql/warehouses and serving-endpoints.

#Identities

for sp in w.service_principals.list():
    print(sp.application_id, sp.display_name, sp.active)

for g in w.groups.list(filter='displayName eq "data-platform-ops"'):
    print(g.id, [m.display for m in (g.members or [])])

With identity federation, manage users and groups at account level (a.users, a.groups) and assign them to workspaces. Workspace-level SCIM changes may conflict with your Entra ID sync.

#Unity Catalog objects

for cat in w.catalogs.list():
    print(cat.name, cat.owner)

for t in w.tables.list(catalog_name="main", schema_name="sales"):
    print(t.full_name, t.table_type, t.owner)

t = w.tables.get(full_name="main.sales.orders")
print(t.storage_location, t.data_source_format)

For grants, running SQL (next section) is usually clearer and more stable across SDK versions than the typed grants API.

#SQL through a warehouse

from databricks.sdk.service.sql import StatementState

resp = w.statement_execution.execute_statement(
    warehouse_id="abcdef1234567890",
    statement="SELECT job_id, count(*) AS runs FROM system.lakeflow.job_run_timeline "
              "WHERE period_start_time > now() - INTERVAL 1 DAY GROUP BY job_id",
    wait_timeout="30s")

if resp.status.state == StatementState.SUCCEEDED:
    cols = [c.name for c in resp.manifest.schema.columns]
    for row in resp.result.data_array or []:
        print(dict(zip(cols, row)))
else:
    print(resp.status.state, resp.status.error)

Large results arrive in chunks. Fetch the rest with w.statement_execution.get_statement_result_chunk_n(statement_id, chunk_index). For grants: statement="GRANT SELECT ON TABLE main.sales.orders TO \sales-analysts`"`.

#Workspace files and secrets

for o in w.workspace.list("/Workspace/Shared", recursive=True):
    print(o.object_type, o.path)

for s in w.secrets.list_scopes():
    print(s.name, s.backend_type)

#Errors, retries and rate limits

from databricks.sdk.core import Config
from databricks.sdk.errors import NotFound, PermissionDenied, DatabricksError

w = WorkspaceClient(config=Config(profile="prod", rate_limit=10, retry_timeout_seconds=600))

try:
    w.jobs.get(job_id=999999)
except NotFound:
    print("job does not exist")
except PermissionDenied as e:
    print("no access:", e)
except DatabricksError as e:
    print(e.error_code, e)

Turn on request logging when a call misbehaves:

import logging
logging.basicConfig(level=logging.DEBUG)
logging.getLogger("databricks.sdk").setLevel(logging.DEBUG)

#Quick reference