Generalize CLI token source into progressive command list

Replace the explicit force_cmd/fallback_cmd fields with a CliCommand
dataclass and an optional commands list on CliTokenSource.

When commands is provided, refresh() delegates to _refresh_progressive()
which walks the list from activeCommandIndex, falling back on
unsupported-flag errors. When commands is None, refresh() delegates to
_refresh_single() which preserves the original fallback behavior
with zero changes for AzureCliTokenSource.

DatabricksCliTokenSource._build_commands() produces the progressive
list: --profile + --force-refresh first, plain --profile second, and
--host as a terminal fallback. --force-refresh is only paired with
--profile, never with --host. Adding future flags (e.g. --scopes)
requires only adding entries to _build_commands().

Author: mihaimitrea-db

NEXT_CHANGELOG.md

@@ -19,6 +19,7 @@
 ### Internal Changes
 * Replace the async-disabling mechanism on token refresh failure with a 1-minute retry backoff. Previously, a single failed async refresh would disable proactive token renewal until the token expired. Now, the SDK waits a short cooldown period and retries, improving resilience to transient errors.
 * Extract `_resolve_profile` to simplify config file loading and improve `__settings__` error messages.
+* Generalize CLI token source into a progressive command list for forward-compatible flag support.
 
 ### API Changes
 * Add `create_catalog()`, `create_synced_table()`, `delete_catalog()`, `delete_synced_table()`, `get_catalog()` and `get_synced_table()` methods for [w.postgres](https://databricks-sdk-py.readthedocs.io/en/latest/workspace/postgres/postgres.html) workspace-level service.

databricks/sdk/credentials_provider.py

@@ -1,5 +1,6 @@
 import abc
 import base64
+import dataclasses
 import functools
 import io
 import json
@@ -649,6 +650,15 @@ def refreshed_headers() -> Dict[str, str]:
     return OAuthCredentialsProvider(refreshed_headers, token)
 
 
+@dataclasses.dataclass
+class CliCommand:
+    """A single CLI command variant with metadata for progressive fallback."""
+
+    args: List[str]
+    flags: List[str]
+    warning: str
+
+
 class CliTokenSource(oauth.Refreshable):
 
     def __init__(
@@ -659,6 +669,7 @@ def __init__(
         expiry_field: str,
         disable_async: bool = True,
         fallback_cmd: Optional[List[str]] = None,
+        commands: Optional[List[CliCommand]] = None,
     ):
         super().__init__(disable_async=disable_async)
         self._cmd = cmd
@@ -670,6 +681,8 @@ def __init__(
         self._token_type_field = token_type_field
         self._access_token_field = access_token_field
         self._expiry_field = expiry_field
+        self._commands = commands
+        self._active_command_index = -1
 
     @staticmethod
     def _parse_expiry(expiry: str) -> datetime:
@@ -700,7 +713,18 @@ def _exec_cli_command(self, cmd: List[str]) -> oauth.Token:
             message = "\n".join(filter(None, [stdout, stderr]))
             raise IOError(f"cannot get access token: {message}") from e
 
+    @staticmethod
+    def _is_unknown_flag_error(error: IOError, flags: List[str]) -> bool:
+        """Check if the error indicates the CLI rejected one of the given flags."""
+        msg = str(error)
+        return any(f"unknown flag: {flag}" in msg for flag in flags)
+
     def refresh(self) -> oauth.Token:
+        if self._commands is not None:
+            return self._refresh_progressive()
+        return self._refresh_single()
+
+    def _refresh_single(self) -> oauth.Token:
         try:
             return self._exec_cli_command(self._cmd)
         except IOError as e:
@@ -712,6 +736,30 @@ def refresh(self) -> oauth.Token:
                 return self._exec_cli_command(self._fallback_cmd)
             raise
 
+    def _refresh_progressive(self) -> oauth.Token:
+        idx = self._active_command_index
+        if idx >= 0:
+            return self._exec_cli_command(self._commands[idx].args)
+        return self._probe_and_exec()
+
+    def _probe_and_exec(self) -> oauth.Token:
+        """Walk the command list to find a CLI command that succeeds.
+
+        When a command fails with "unknown flag" for one of its flags, log a
+        warning and try the next.  On success, store _active_command_index so
+        future calls skip probing.
+        """
+        for i, cmd in enumerate(self._commands):
+            try:
+                token = self._exec_cli_command(cmd.args)
+                self._active_command_index = i
+                return token
+            except IOError as e:
+                is_last = i == len(self._commands) - 1
+                if is_last or not self._is_unknown_flag_error(e, cmd.flags):
+                    raise
+                logger.warning(cmd.warning)
+
 
 def _run_subprocess(
     popenargs,
@@ -899,15 +947,7 @@ def __init__(self, cfg: "Config"):
         elif cli_path.count("/") == 0:
             cli_path = self.__class__._find_executable(cli_path)
 
-        fallback_cmd = None
-        self._force_cmd = None
-        if cfg.profile:
-            args = ["auth", "token", "--profile", cfg.profile]
-            self._force_cmd = [cli_path, *args, "--force-refresh"]
-            if cfg.host:
-                fallback_cmd = [cli_path, *self.__class__._build_host_args(cfg)]
-        else:
-            args = self.__class__._build_host_args(cfg)
+        commands = self.__class__._build_commands(cli_path, cfg)
 
         # get_scopes() defaults to ["all-apis"] when nothing is configured, which would
         # cause false-positive mismatches against every token that wasn't issued with
@@ -917,30 +957,57 @@ def __init__(self, cfg: "Config"):
         self._host = cfg.host
 
         super().__init__(
-            cmd=[cli_path, *args],
+            cmd=commands[-1].args,
             token_type_field="token_type",
             access_token_field="access_token",
             expiry_field="expiry",
             disable_async=cfg.disable_async_token_refresh,
-            fallback_cmd=fallback_cmd,
+            commands=commands,
         )
 
-    def refresh(self) -> oauth.Token:
-        if self._force_cmd is None:
-            token = super().refresh()
-        else:
-            try:
-                token = self._exec_cli_command(self._force_cmd)
-            except IOError as e:
-                err_msg = str(e)
-                if "unknown flag: --force-refresh" in err_msg or "unknown flag: --profile" in err_msg:
-                    logger.warning(
-                        "Databricks CLI does not support --force-refresh. "
-                        "Please upgrade your CLI to the latest version."
+    @staticmethod
+    def _build_commands(cli_path: str, cfg: "Config") -> List[CliCommand]:
+        commands: List[CliCommand] = []
+        if cfg.profile:
+            profile_args = [cli_path, "auth", "token", "--profile", cfg.profile]
+            commands.append(
+                CliCommand(
+                    args=profile_args + ["--force-refresh"],
+                    flags=["--force-refresh", "--profile"],
+                    warning="Databricks CLI does not support --force-refresh. "
+                    "Please upgrade your CLI to the latest version.",
+                )
+            )
+            commands.append(
+                CliCommand(
+                    args=profile_args,
+                    flags=["--profile"],
+                    warning="Databricks CLI does not support --profile flag. "
+                    "Falling back to --host. "
+                    "Please upgrade your CLI to the latest version.",
+                )
+            )
+            if cfg.host:
+                commands.append(
+                    CliCommand(
+                        args=[cli_path, *DatabricksCliTokenSource._build_host_args(cfg)],
+                        flags=[],
+                        warning="",
                     )
-                    token = super().refresh()
-                else:
-                    raise
+                )
+        else:
+            host_args = [cli_path, *DatabricksCliTokenSource._build_host_args(cfg)]
+            commands.append(
+                CliCommand(
+                    args=host_args,
+                    flags=[],
+                    warning="",
+                )
+            )
+        return commands
+
+    def refresh(self) -> oauth.Token:
+        token = super().refresh()
         if self._requested_scopes:
             self._validate_token_scopes(token)
         return token