[Feature] Support custom auth method

[pradipsodha-acceleratedata]

Following up on the closed PR, and aware of

• recent major cleanups
• made more robust and stable (and that's still priority)

and it would be selfish 😅 to ask for such new feature but we really want that feature and we are open to contribute and also happy if project has other priorities first then this.

---

Context

• I'm running dbt-fabricspark from a desktop tool that already has the user logged in via its own oauth flow.
• The tool exposes a small local helper that returns fresh Fabric access tokens on demand.
• With the current auth methods (CLI, SPN, fabric_notebook), there's no way to plug into the tokens the tool already has, so the user ends up logging in twice (once in the tool, once via az login) and dbt uses the second one.

Same shape pops up in a couple of other places:

• CI pipelines using Entra Workload Identity Federation (GitHub Actions, Azure DevOps). The runner gets a short-lived federated token, nothing in the current adapter consumes it.
• Vault / corporate SSO proxies that mint short-lived Azure tokens.
• Any flow where someone has a valid token but doesn't have one of the three credential classes the adapter knows about.

Possible solutions

ENV base auth

inspire by fab cli [FAB_TOKEN], where read token from env variable and any other flow get token however it want to derive and set env variable, it's easy and fast to implement but it's not seamless approach in POV, cause it's nature of env variable, once dbt job trigger and then if token expires (mostly az token has 1 hour expiry) then job will fail and also other flow needs to handle refresh token part also.

token_credential

https://learn.microsoft.com/en-us/python/api/azure-core/azure.core.credentials.tokencredential?view=azure-python

dev:
  type: fabricspark
  method: livy
  authentication: token_credential
  credential_class: "my_company.auth.ExternalTokenCredential"
  credential_kwargs:
    token_url: "{{ env_var('TOKEN_URL') }}"
    user_id: "{{ env_var('USER_ID') }}"

token_command

authentication: token_command
token_command:
  - vault
  - read
  - -format=json
  - fabric/creds/dbt
token_path: data.access_token         # default: "access_token"
expires_path: data.lease_duration     # default: "expires_in"

inspire by:

• AWS CLI's credential_process
• gcloud's credential_source
• git's credential helpers

external_token

dev:
  type: fabricspark
  method: livy
  authentication: external_token

  external_token_url: "{{ env_var('TOKEN_URL') }}"
  external_token_params:
    user_id: "{{ env_var('USER_ID') }}"

  # external_token_headers:
  #   X-Bootstrap-Token: "{{ env_var('BOOTSTRAP_TOKEN') }}"

  # Optional response shape. Override for non-standard brokers.
  # external_token_response_path: "access_token"
  # external_token_expires_in_path: "expires_in"

[pradipsodha-acceleratedata]

Raki (@mdrakiburrahman) 😅 As you suggested added token_credential

[Raki (mdrakiburrahman)]

Thank you!

How would this get invoked?

authentication: token_credential
  credential_class: "my_company.auth.ExternalTokenCredential"

Would you use reflection in Python?

Forget dbt, can you share a standalone python script where ExternalTokenCredential is invoked, if that is not defined in that python script's codebase, but is available in the venv?

dbt is a CLI tool, so it'd have to grab this custom class and invoke it via reflection in the token refresh callback.

I'm curious to see a E2E POC of this working. If that works wiring this up in the adapter should be straightforward.

[pradipsodha-acceleratedata]

Would you use reflection in Python?

yes yes, just like Django uses for AUTHENTICATION_BACKENDS (dotted path -> importlib.import_string -> duck-typed against an interface), airflow does the same thing for its auth backends.

Between between dotted path (my_company.auth.ExternalTokenCredential) vs entry-points (which i guess dbt-core uses), i lean toward dotted path cause for one evn there can be one token provider and doesn't needs to kknow at runtime.

Forget dbt, can you share a standalone python script where ExternalTokenCredential is invoked, if that is not defined in that python script's codebase, but is available in the venv?

Cred package:

custom_creds/
├── pyproject.toml
└── custom_creds/
    ├── __init__.py
    └── auth.py

custom_creds/auth.py

import time
from azure.core.credentials import AccessToken

class ExternalTokenCredential:
    def __init__(self, token_url, user_id):
        self.token_url = token_url
        self.user_id = user_id

    def get_token(self, *scopes, **kwargs) -> AccessToken:
        token = f"fake-token-for-{self.user_id}-scope-{scopes[0]}"
        expires_on = int(time.time()) + 3600
        return AccessToken(token, expires_on)

standalone python script (ie dbt)

import importlib

def load_credential(dotted_path: str, kwargs: dict):
    module_path, _, class_name = dotted_path.rpartition(".")
    cls = getattr(importlib.import_module(module_path), class_name)
    instance = cls(**kwargs)
    if not hasattr(instance, "get_token"):
        raise TypeError(
            f"{dotted_path} does not implement TokenCredential (missing get_token)"
        )
    return instance

# Simulating what would come from profiles.yml
profile = {
    "credential_class": "custom_creds.auth.ExternalTokenCredential",
    "credential_kwargs": {
        "token_url": "https://broker.internal/token",
        "user_id": "alice",
    },
}

cred = load_credential(profile["credential_class"], profile["credential_kwargs"])
token = cred.get_token("https://api.fabric.microsoft.com/.default")
print(f"token={token.token[:30]}... expires_on={token.expires_on}")

full POC: https://github.com/pradipsodha-acceleratedata/external-token-cred-poc and instead of using duck-typing can use isinstance(instance, TokenCredential)).

[Raki (mdrakiburrahman)]

Very interesting.

I wasn't aware this was a pattern in Python ecosystem already.

Are you comfortable opening a PR?
Please note that it'd be good if you can run the CI tests locally in your own Fabric instance, it'll significantly help get it merged faster, because on failures you will not be able to rerun CI yourself on our Fabric workspace.

See here on how to setup a devcontainer so you can locally test with your AI etc in your personal Fabric, there's a demo video as well:

https://github.com/microsoft/dbt-fabricspark/tree/main/contrib

LMK if you run into any issues with the dev env setup, happy to help!

[pradipsodha-acceleratedata]

Awesome 😃, thanks 🙌 Yes, happy to open PR, i will link this discussion in the PR.

Please note that it'd be good if you can run the CI tests locally in your own Fabric instance, it'll significantly help get it merged faster, because on failures you will not be able to rerun CI yourself on our Fabric workspace.

sure sure, will tests locally on my fabric instance.

LMK if you run into any issues with the dev env setup, happy to help!

💙

[pradipsodha-acceleratedata]

#177