# How do I assign CAN_MANAGE to all my compute resources in Databricks?

URL: https://select.dev/docs/reference/faq/automated-savings/how-do-i-assign-can_manage-to-all-my-compute-resources-in-databricks
Index of all SELECT docs: https://select.dev/docs/llms.txt

> An automated way to provision the required access needed for Databricks Automated Savings

This script helps provision the required access in an automated manner.

```python
# Dependencies: pip install databricks-sdk

# ── Config — edit these before running ───────────────────────────────────────

WORKSPACE_URL = "<https://dbc-xxxxx.cloud.databricks.com>"
ACCOUNT_ID = "<databricks-account-id>"
SERVICE_PRINCIPAL_NAME = "<service-principal-name>"

# ─────────────────────────────────────────────────────────────────────────────

from urllib.parse import urlparse

from databricks.sdk import AccountClient
from databricks.sdk.service.compute import ClusterSource
from databricks.sdk.service.iam import (
    AccessControlRequest,
    PermissionLevel,
    WorkspacePermission,
)


def main():
    parsed = urlparse(WORKSPACE_URL)
    workspace_host = f"{parsed.scheme}://{parsed.netloc}"
    account_host = detect_account_host(workspace_host)
    account_client = AccountClient(
        host=account_host, account_id=ACCOUNT_ID, auth_type="external-browser"
    )

    ws_info = find_current_workspace(account_client, workspace_host)
    region = ws_info.location or ws_info.aws_region
    print(f"Workspace : {ws_info.workspace_name}")
    print(f"Region    : {region}")

    sp = get_service_principal(account_client, SERVICE_PRINCIPAL_NAME)

    workspaces_in_region = get_workspaces_in_region(account_client, region)
    print(f"\n{len(workspaces_in_region)} workspace(s) in region '{region}'")

    for ws in workspaces_in_region:
        print(f"\n  Workspace: {ws.workspace_name}")
        ws_client = account_client.get_workspace_client(ws)

        ensure_sp_assigned_to_workspace(account_client, ws_client, ws, sp)

        print(f"  Granting CAN_MANAGE to SP on jobs in {ws.workspace_name}")
        for job in ws_client.jobs.list():
            grant_permission(
                ws_client, "jobs", job.job_id, sp, PermissionLevel.CAN_MANAGE
            )

        print(f"  Granting CAN_MANAGE to SP on clusters in {ws.workspace_name}")
        for cluster in ws_client.clusters.list():
            if cluster.cluster_source not in (ClusterSource.API, ClusterSource.UI):
                continue
            grant_permission(
                ws_client,
                "clusters",
                cluster.cluster_id,
                sp,
                PermissionLevel.CAN_MANAGE,
            )

        print(f"  Granting CAN_MANAGE to SP on warehouses in {ws.workspace_name}")
        for wh in ws_client.warehouses.list():
            grant_permission(
                ws_client, "warehouses", wh.id, sp, PermissionLevel.CAN_MANAGE
            )

    print(
        "\n── Done ─────────────────────────────────────────────────────────────────────"
    )
    print(f"  Account ID   : {ACCOUNT_ID}")
    print(f"  SP Client ID : {sp.application_id}")
    print(
        "─────────────────────────────────────────────────────────────────────────────"
    )


# ── Helpers ───────────────────────────────────────────────────────────────────


def detect_account_host(workspace_host):
    if "azuredatabricks.net" in workspace_host:
        return "https://accounts.azuredatabricks.net"
    elif "gcp.databricks.com" in workspace_host:
        return "https://accounts.gcp.databricks.com"
    else:
        return "https://accounts.cloud.databricks.com"


def find_current_workspace(account_client, workspace_host):
    for ws in account_client.workspaces.list():
        if ws.deployment_name and ws.deployment_name in workspace_host:
            return ws
    raise ValueError(
        f"No workspace found matching '{workspace_host}'. "
        "Ensure WORKSPACE_URL contains a valid Databricks workspace URL."
    )


def get_service_principal(account_client, name):
    for s in account_client.service_principals.list():
        if s.display_name == name:
            print(f"\nFound SP '{s.display_name}' (app_id={s.application_id})")
            return s
    raise ValueError(
        f"No service principal named '{name}' found. "
        "This script does not create one — provision it first."
    )


def get_workspaces_in_region(account_client, region):
    workspaces = []
    for ws in account_client.workspaces.list():
        if (ws.location or ws.aws_region) == region:
            workspaces.append(ws)
    return workspaces


def ensure_sp_assigned_to_workspace(account_client, ws_client, ws, sp):
    for s in ws_client.service_principals.list():
        if s.application_id == sp.application_id:
            return
    print("    Assigning SP to workspace …")
    account_client.workspace_assignment.update(
        workspace_id=ws.workspace_id,
        principal_id=int(sp.id),
        permissions=[WorkspacePermission.USER],
    )


def grant_permission(ws_client, resource_type, resource_id, sp, level):
    sp_app_id = str(sp.application_id)
    perms = ws_client.permissions.get(resource_type, resource_id)
    already_granted = any(
        entry.service_principal_name == sp_app_id
        and any(p.permission_level == level for p in (entry.all_permissions or []))
        for entry in (perms.access_control_list or [])
    )
    if already_granted:
        print(
            f"      {level.value} on {resource_type}/{resource_id} already granted, skipping."
        )
        return
    print(f"      Granting {level.value} on {resource_type}/{resource_id} …")
    ws_client.permissions.update(
        resource_type,
        resource_id,
        access_control_list=[
            AccessControlRequest(
                service_principal_name=sp_app_id, permission_level=level
            )
        ],
    )


if __name__ == "__main__":
    main()
```