feat(mssql): support injectable access_token and azure_credential

dlt/destinations/impl/fabric/configuration.py

@@ -1,23 +1,43 @@
 """Configuration for Fabric Warehouse destination - extends Synapse configuration with COPY INTO support"""
 
 from typing import Optional, Final, ClassVar, Dict, Any, List
-from dlt.common.configuration import configspec
+from dlt.common.configuration import configspec, NotResolved
 from dlt.common.configuration.specs import AzureServicePrincipalCredentials
 from dlt.common.destination.client import DestinationClientDwhWithStagingConfiguration
-from dlt.common.exceptions import MissingDependencyException
+from dlt.common.typing import TSecretStrValue, Annotated
 from dlt.common.utils import digest128
-from dlt import version
+from dlt.destinations.impl.mssql.configuration import (
+    apply_authentication_to_dsn,
+    build_token_attrs_before,
+    setup_token_credential,
+    validate_authentication,
+)
 
-_AZURE_STORAGE_EXTRA = f"{version.DLT_PKG_NAME}[az]"
+# Fabric Warehouse only supports Entra ID authentication, so it defaults to Service Principal
+# (the shared MsSql auth machinery additionally enables azure-identity token methods).
+_DEFAULT_AUTHENTICATION = "ActiveDirectoryServicePrincipal"
 
 
 @configspec(init=False)
 class FabricCredentials(AzureServicePrincipalCredentials):
-    """Credentials for Microsoft Fabric Warehouse with Service Principal authentication.
+    """Credentials for Microsoft Fabric Warehouse.
 
-    Fabric Warehouse requires Azure AD Service Principal authentication.
-    Inherits from AzureServicePrincipalCredentials for Service Principal fields and
-    automatic fallback to DefaultAzureCredential.
+    Supports several Entra ID authentication methods, selected through `authentication`:
+
+    * **Driver-native** (the ODBC driver authenticates): `ActiveDirectoryServicePrincipal`
+      (default), `ActiveDirectoryPassword`, `ActiveDirectoryIntegrated`,
+      `ActiveDirectoryInteractive`, `ActiveDirectoryMsi`.
+    * **azure-identity** (dlt acquires an access token and injects it, works cross-platform):
+      `ActiveDirectoryDefault` (alias `default`, uses `DefaultAzureCredential`),
+      `ActiveDirectoryDeviceCode` (uses `DeviceCodeCredential`).
+
+    When `authentication` is left at its default but no Service Principal secret is configured,
+    dlt falls back to `ActiveDirectoryDefault` and injects its token.
+
+    Alternatively, `access_token` or `azure_credential` can be injected directly, bypassing
+    `authentication` entirely (see their docstrings for precedence).
+
+    Inherits from AzureServicePrincipalCredentials for the Service Principal fields.
     """
 
     drivername: str = "mssql+pyodbc"
@@ -35,43 +55,65 @@ class FabricCredentials(AzureServicePrincipalCredentials):
     connect_timeout: int = 15
     """Connection timeout in seconds (default: 15)"""
 
+    authentication: str = _DEFAULT_AUTHENTICATION
+    """Authentication method. Driver-native: `ActiveDirectoryServicePrincipal` (default),
+    `ActiveDirectoryPassword`, `ActiveDirectoryIntegrated`, `ActiveDirectoryInteractive`,
+    `ActiveDirectoryMsi`. azure-identity (token injected by dlt): `ActiveDirectoryDefault`
+    (alias `default`), `ActiveDirectoryDeviceCode`."""
+
+    username: str | None = None
+    """User principal name, used with `ActiveDirectoryPassword` authentication."""
+
+    password: TSecretStrValue | None = None
+    """Password, used with `ActiveDirectoryPassword` authentication."""
+
+    access_token: Optional[TSecretStrValue] = None
+    """Pre-acquired Entra ID access token. When set, dlt injects it directly via `attrs_before`
+    without acquiring anything. Takes precedence over `azure_credential` and `authentication`."""
+
+    azure_credential: Annotated[Optional[Any], NotResolved()] = None
+    """An externally constructed `azure.core.credentials.TokenCredential` (e.g.
+    `DefaultAzureCredential()`) injected at runtime, not resolved from config providers. dlt
+    calls its `get_token()` to acquire an access token. Takes precedence over `authentication`,
+    but `access_token` takes precedence over this."""
+
     # Override to make optional - not needed for Fabric Warehouse credentials (only for staging)
     azure_storage_account_name: Optional[str] = None
     """Not used for Fabric Warehouse credentials (only staging credentials need this)"""
 
     def on_partial(self) -> None:
-        """Enable fallback to DefaultAzureCredential if explicit credentials not provided."""
-        try:
-            from azure.identity import DefaultAzureCredential
-        except ModuleNotFoundError:
-            raise MissingDependencyException(self.__class__.__name__, [_AZURE_STORAGE_EXTRA])
-
-        # If no explicit Service Principal credentials, use default credentials
-        if not self.azure_client_id or not self.azure_client_secret or not self.azure_tenant_id:
-            self._set_default_credentials(DefaultAzureCredential())
-            # Resolve if we have warehouse connection details (not storage account name)
-            if self.host and self.database:
-                self.resolve()
+        """Set up token-based credentials and resolve once host and database are known.
+
+        Token-based methods (and the default Service Principal method without a secret) get an
+        azure-identity credential whose token is injected into the connection. Driver-native
+        methods need no token and resolve as-is. Auth logic is shared with `MsSqlCredentials`.
+        """
+        setup_token_credential(self)
+        # Resolve if we have the warehouse connection details (not the storage account name)
+        if self.host and self.database:
+            self.resolve()
+
+    def on_resolved(self) -> None:
+        """Validate the configured authentication method."""
+        validate_authentication(self)
 
     def get_odbc_dsn_dict(self) -> Dict[str, Any]:
         """Build ODBC DSN dictionary with Fabric-specific settings."""
-        params = {
+        params: dict[str, Any] = {
             "DRIVER": "{ODBC Driver 18 for SQL Server}",
             "SERVER": f"{self.host},{self.port}",
             "DATABASE": self.database,
-            "AUTHENTICATION": "ActiveDirectoryServicePrincipal",
             "LongAsMax": "yes",  # Required for UTF-8 collation support
             "Encrypt": "yes",
             "TrustServerCertificate": "no",
         }
-
-        # Add Service Principal credentials if provided
-        if self.azure_client_id and self.azure_tenant_id and self.azure_client_secret:
-            params["UID"] = f"{self.azure_client_id}@{self.azure_tenant_id}"
-            params["PWD"] = str(self.azure_client_secret)
-
+        apply_authentication_to_dsn(self, params)
         return params
 
+    def to_odbc_attrs_before(self) -> dict[int, bytes] | None:
+        """Return pyodbc `attrs_before` with an Entra ID access token, or None for driver-native auth."""
+        return build_token_attrs_before(self)
+
     def to_odbc_dsn(self) -> str:
         """Build ODBC connection string for pyodbc."""
         params = self.get_odbc_dsn_dict()

dlt/destinations/impl/fabric/fabric.py

@@ -107,6 +107,12 @@ def _ensure_fabric_token_initialized(
         if cache_key in self._token_initialized_cache:
             return
 
+        if not credentials.azure_client_secret:
+            # No Service Principal secret configured: ClientSecretCredential would raise before
+            # any data moves. Skip the proactive Fabric token initialization rather than fail the
+            # whole load; COPY INTO still works as long as the staging credential is otherwise valid.
+            return
+
         try:
             import requests
             from azure.identity import ClientSecretCredential

dlt/destinations/impl/fabric/sql_client.py

@@ -13,6 +13,8 @@ class FabricSqlClient(SynapseSqlClient):
     """SQL client for Microsoft Fabric Warehouse
 
     Inherits all behavior from Synapse since Fabric Warehouse is built on Synapse technology.
+    Azure AD token injection (for token-based authentication) is handled by the shared
+    `PyOdbcMsSqlClient.open_connection`.
     """
 
     def __init__(