From dfa5a5c09253c0bfd8a10faa630c307b11a87674 Mon Sep 17 00:00:00 2001
From: karunpoudel <62040859+karunpoudel@users.noreply.github.com>
Date: Tue, 24 Jun 2025 00:17:09 -0400
Subject: [PATCH 01/18] init

---
 .../microsoft/azure/hooks/base_azure.py       | 84 +++++++++++++++----
 1 file changed, 68 insertions(+), 16 deletions(-)

diff --git a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
index fcc01248b755a..8301f0c566b06 100644
--- a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
+++ b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
@@ -16,16 +16,19 @@
 # under the License.
 from __future__ import annotations
 
-from typing import Any
+from typing import Any, Optional
 
 from azure.common.client_factory import get_client_from_auth_file, get_client_from_json_dict
 from azure.common.credentials import ServicePrincipalCredentials
+from azure.identity import ClientSecretCredential, DefaultAzureCredential
 
 from airflow.exceptions import AirflowException
 from airflow.hooks.base import BaseHook
+from airflow.models import Connection
 from airflow.providers.microsoft.azure.utils import (
     AzureIdentityCredentialAdapter,
     add_managed_identity_connection_widgets,
+    get_sync_default_azure_credential,
 )
 
 
@@ -85,7 +88,7 @@ def get_ui_field_behaviour(cls) -> dict[str, Any]:
             },
         }
 
-    def __init__(self, sdk_client: Any, conn_id: str = "azure_default"):
+    def __init__(self, sdk_client: Any = None, conn_id: str = "azure_default"):
         self.sdk_client = sdk_client
         self.conn_id = conn_id
         super().__init__()
@@ -96,6 +99,10 @@ def get_conn(self) -> Any:
 
         :return: the authenticated client.
         """
+        if not self.sdk_client:
+            raise ValueError(
+                "`sdk_client` must be provided to AzureBaseHook to use `get_conn` method."
+            )
         conn = self.get_connection(self.conn_id)
         tenant = conn.extra_dejson.get("tenantId")
         subscription_id = conn.extra_dejson.get("subscriptionId")
@@ -111,22 +118,67 @@ def get_conn(self) -> Any:
             self.log.info("Getting connection using a JSON config.")
             return get_client_from_json_dict(client_class=self.sdk_client, config_dict=key_json)
 
-        credentials: ServicePrincipalCredentials | AzureIdentityCredentialAdapter
-        if all([conn.login, conn.password, tenant]):
-            self.log.info("Getting connection using specific credentials and subscription_id.")
-            credentials = ServicePrincipalCredentials(
-                client_id=conn.login, secret=conn.password, tenant=tenant
-            )
-        else:
-            self.log.info("Using DefaultAzureCredential as credential")
-            managed_identity_client_id = conn.extra_dejson.get("managed_identity_client_id")
-            workload_identity_tenant_id = conn.extra_dejson.get("workload_identity_tenant_id")
-            credentials = AzureIdentityCredentialAdapter(
-                managed_identity_client_id=managed_identity_client_id,
-                workload_identity_tenant_id=workload_identity_tenant_id,
-            )
+        credentials = self.get_credentials(conn=conn)
 
         return self.sdk_client(
             credentials=credentials,
             subscription_id=subscription_id,
         )
+
+    def get_credentials(
+        self, *, conn: Optional[Connection] = None
+    ) -> ServicePrincipalCredentials | AzureIdentityCredentialAdapter | ClientSecretCredential | DefaultAzureCredential:
+        """
+        Get Azure credentials object for the connection.
+
+        Azure Identity based credentials object can be used to get token.
+        Older Credential object are supported for backward compatibility.
+
+        :return: The Azure credentials object
+        """
+        if not conn:
+            conn = self.get_connection(self.conn_id)
+        tenant = conn.extra_dejson.get("tenantId")
+        if not tenant and conn.extra_dejson.get("extra__azure__tenantId"):
+            warnings.warn(
+                "`extra__azure__tenantId` is deprecated in azure connection extra, "
+                "please use `tenantId` instead",
+                AirflowProviderDeprecationWarning,
+                stacklevel=2,
+            )
+            tenant = conn.extra_dejson.get("extra__azure__tenantId")
+        use_azure_identity_creds_object = conn.extra_dejson.get(
+            "use_azure_identity_creds_object", False
+        )
+        credentials: ServicePrincipalCredentials | AzureIdentityCredentialAdapter | ClientSecretCredential | DefaultAzureCredential
+        if all([conn.login, conn.password, tenant]):
+            self.log.info(
+                "Getting credentials using specific credentials and subscription_id."
+            )
+            if use_azure_identity_creds_object:
+                credentials = ClientSecretCredential(
+                    client_id=conn.login, client_secret=conn.password, tenant_id=tenant
+                )
+            else:
+                credentials = ServicePrincipalCredentials(
+                    client_id=conn.login, secret=conn.password, tenant=tenant
+                )
+        else:
+            self.log.info("Using DefaultAzureCredential as credential")
+            managed_identity_client_id = conn.extra_dejson.get(
+                "managed_identity_client_id"
+            )
+            workload_identity_tenant_id = conn.extra_dejson.get(
+                "workload_identity_tenant_id"
+            )
+            if use_azure_identity_creds_object:
+                credentials = get_sync_default_azure_credential(
+                    managed_identity_client_id=managed_identity_client_id,
+                    workload_identity_tenant_id=workload_identity_tenant_id,
+                )
+            else:
+                credentials = AzureIdentityCredentialAdapter(
+                    managed_identity_client_id=managed_identity_client_id,
+                    workload_identity_tenant_id=workload_identity_tenant_id,
+                )
+        return credentials

From 56dae4dc33856250ae9e4b920af95520d181c2c7 Mon Sep 17 00:00:00 2001
From: Karun Poudel <karun.poudel@chrobinson.com>
Date: Tue, 24 Jun 2025 17:38:04 +0000
Subject: [PATCH 02/18] fx

---
 .../microsoft/azure/hooks/base_azure.py       | 56 +++++++--------
 .../microsoft/azure/hooks/test_base_azure.py  | 71 ++++++++++++++++++-
 2 files changed, 98 insertions(+), 29 deletions(-)

diff --git a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
index 8301f0c566b06..5705dcbebed67 100644
--- a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
+++ b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
@@ -100,9 +100,7 @@ def get_conn(self) -> Any:
         :return: the authenticated client.
         """
         if not self.sdk_client:
-            raise ValueError(
-                "`sdk_client` must be provided to AzureBaseHook to use `get_conn` method."
-            )
+            raise ValueError("`sdk_client` must be provided to AzureBaseHook to use `get_conn` method.")
         conn = self.get_connection(self.conn_id)
         tenant = conn.extra_dejson.get("tenantId")
         subscription_id = conn.extra_dejson.get("subscriptionId")
@@ -118,23 +116,28 @@ def get_conn(self) -> Any:
             self.log.info("Getting connection using a JSON config.")
             return get_client_from_json_dict(client_class=self.sdk_client, config_dict=key_json)
 
-        credentials = self.get_credentials(conn=conn)
+        credentials = self.get_credential(conn=conn)
 
         return self.sdk_client(
             credentials=credentials,
             subscription_id=subscription_id,
         )
 
-    def get_credentials(
+    def get_credential(
         self, *, conn: Optional[Connection] = None
-    ) -> ServicePrincipalCredentials | AzureIdentityCredentialAdapter | ClientSecretCredential | DefaultAzureCredential:
+    ) -> (
+        ServicePrincipalCredentials
+        | AzureIdentityCredentialAdapter
+        | ClientSecretCredential
+        | DefaultAzureCredential
+    ):
         """
-        Get Azure credentials object for the connection.
+        Get Azure credential object for the connection.
 
-        Azure Identity based credentials object can be used to get token.
+        Azure Identity based credential object can be used to get token.
         Older Credential object are supported for backward compatibility.
 
-        :return: The Azure credentials object
+        :return: The Azure credential object
         """
         if not conn:
             conn = self.get_connection(self.conn_id)
@@ -147,38 +150,35 @@ def get_credentials(
                 stacklevel=2,
             )
             tenant = conn.extra_dejson.get("extra__azure__tenantId")
-        use_azure_identity_creds_object = conn.extra_dejson.get(
-            "use_azure_identity_creds_object", False
+        use_azure_identity_object = conn.extra_dejson.get("use_azure_identity_object", False)
+        credential: (
+            ServicePrincipalCredentials
+            | AzureIdentityCredentialAdapter
+            | ClientSecretCredential
+            | DefaultAzureCredential
         )
-        credentials: ServicePrincipalCredentials | AzureIdentityCredentialAdapter | ClientSecretCredential | DefaultAzureCredential
         if all([conn.login, conn.password, tenant]):
-            self.log.info(
-                "Getting credentials using specific credentials and subscription_id."
-            )
-            if use_azure_identity_creds_object:
-                credentials = ClientSecretCredential(
+            self.log.info("Getting credentials using specific credentials and subscription_id.")
+            if use_azure_identity_object:
+                credential = ClientSecretCredential(
                     client_id=conn.login, client_secret=conn.password, tenant_id=tenant
                 )
             else:
-                credentials = ServicePrincipalCredentials(
+                credential = ServicePrincipalCredentials(
                     client_id=conn.login, secret=conn.password, tenant=tenant
                 )
         else:
             self.log.info("Using DefaultAzureCredential as credential")
-            managed_identity_client_id = conn.extra_dejson.get(
-                "managed_identity_client_id"
-            )
-            workload_identity_tenant_id = conn.extra_dejson.get(
-                "workload_identity_tenant_id"
-            )
-            if use_azure_identity_creds_object:
-                credentials = get_sync_default_azure_credential(
+            managed_identity_client_id = conn.extra_dejson.get("managed_identity_client_id")
+            workload_identity_tenant_id = conn.extra_dejson.get("workload_identity_tenant_id")
+            if use_azure_identity_object:
+                credential = get_sync_default_azure_credential(
                     managed_identity_client_id=managed_identity_client_id,
                     workload_identity_tenant_id=workload_identity_tenant_id,
                 )
             else:
-                credentials = AzureIdentityCredentialAdapter(
+                credential = AzureIdentityCredentialAdapter(
                     managed_identity_client_id=managed_identity_client_id,
                     workload_identity_tenant_id=workload_identity_tenant_id,
                 )
-        return credentials
+        return credential
diff --git a/providers/microsoft/azure/tests/unit/microsoft/azure/hooks/test_base_azure.py b/providers/microsoft/azure/tests/unit/microsoft/azure/hooks/test_base_azure.py
index 89881eae16513..aafcac670059a 100644
--- a/providers/microsoft/azure/tests/unit/microsoft/azure/hooks/test_base_azure.py
+++ b/providers/microsoft/azure/tests/unit/microsoft/azure/hooks/test_base_azure.py
@@ -31,6 +31,7 @@
     Connection = MagicMock()  # type: ignore[misc]
 
 MODULE = "airflow.providers.microsoft.azure.hooks.base_azure"
+UTILS = "airflow.providers.microsoft.azure.utils"
 
 
 class TestBaseAzureHook:
@@ -111,7 +112,7 @@ def test_get_conn_with_credentials(self, mock_spc, mocked_connection):
         indirect=True,
     )
     @patch("azure.common.credentials.ServicePrincipalCredentials")
-    @patch("airflow.providers.microsoft.azure.hooks.base_azure.AzureIdentityCredentialAdapter")
+    @patch(f"{MODULE}.AzureIdentityCredentialAdapter")
     def test_get_conn_fallback_to_azure_identity_credential_adapter(
         self,
         mock_credential_adapter,
@@ -133,3 +134,71 @@ def test_get_conn_fallback_to_azure_identity_credential_adapter(
             credentials=mock_credential,
             subscription_id="subscription_id",
         )
+
+    @patch(f"{MODULE}.ClientSecretCredential")
+    @pytest.mark.parametrize(
+        "mocked_connection",
+        [
+            Connection(
+                conn_id="azure_default",
+                login="my_login",
+                password="my_password",
+                extra={"tenantId": "my_tenant", "use_azure_identity_object": True},
+            ),
+        ],
+        indirect=True,
+    )
+    def test_get_credential_with_client_secret(self, mock_spc, mocked_connection):
+        mock_spc.return_value = "foo-bar"
+        cred = AzureBaseHook().get_credential()
+
+        mock_spc.assert_called_once_with(
+            client_id=mocked_connection.login,
+            client_secret=mocked_connection.password,
+            tenant_id=mocked_connection.extra_dejson["tenantId"],
+        )
+        assert cred == "foo-bar"
+
+    @patch(f"{UTILS}.DefaultAzureCredential")
+    @pytest.mark.parametrize(
+        "mocked_connection",
+        [
+            Connection(
+                conn_id="azure_default",
+                extra={"use_azure_identity_object": True},
+            ),
+        ],
+        indirect=True,
+    )
+    def test_get_credential_with_azure_default_credential(self, mock_spc, mocked_connection):
+        mock_spc.return_value = "foo-bar"
+        cred = AzureBaseHook().get_credential()
+
+        mock_spc.assert_called_once_with()
+        assert cred == "foo-bar"
+
+    @patch(f"{UTILS}.DefaultAzureCredential")
+    @pytest.mark.parametrize(
+        "mocked_connection",
+        [
+            Connection(
+                conn_id="azure_default",
+                extra={
+                    "managed_identity_client_id": "test_client_id",
+                    "workload_identity_tenant_id": "test_tenant_id",
+                    "use_azure_identity_object": True,
+                },
+            ),
+        ],
+        indirect=True,
+    )
+    def test_get_credential_with_azure_default_credential_with_extra(self, mock_spc, mocked_connection):
+        mock_spc.return_value = "foo-bar"
+        cred = AzureBaseHook().get_credential()
+
+        mock_spc.assert_called_once_with(
+            managed_identity_client_id=mocked_connection.extra_dejson.get("managed_identity_client_id"),
+            workload_identity_tenant_id=mocked_connection.extra_dejson.get("workload_identity_tenant_id"),
+            additionally_allowed_tenants=[mocked_connection.extra_dejson.get("workload_identity_tenant_id")],
+        )
+        assert cred == "foo-bar"

From c841b59c9a30d127582ffdcc537b0095d42ee894 Mon Sep 17 00:00:00 2001
From: Karun Poudel <karun.poudel@chrobinson.com>
Date: Tue, 24 Jun 2025 18:07:11 +0000
Subject: [PATCH 03/18] fx

---
 providers/microsoft/azure/docs/connections/azure.rst          | 2 ++
 .../src/airflow/providers/microsoft/azure/hooks/base_azure.py | 4 ++--
 2 files changed, 4 insertions(+), 2 deletions(-)

diff --git a/providers/microsoft/azure/docs/connections/azure.rst b/providers/microsoft/azure/docs/connections/azure.rst
index f8d111fd34a10..79763c9f4cee0 100644
--- a/providers/microsoft/azure/docs/connections/azure.rst
+++ b/providers/microsoft/azure/docs/connections/azure.rst
@@ -74,6 +74,8 @@ Extra (optional)
       It specifies the json that contains the authentication information.
     * ``managed_identity_client_id``:  The client ID of a user-assigned managed identity. If provided with ``workload_identity_tenant_id``, they'll pass to DefaultAzureCredential_.
     * ``workload_identity_tenant_id``: ID of the application's Microsoft Entra tenant. Also called its "directory" ID. If provided with ``managed_identity_client_id``, they'll pass to DefaultAzureCredential_.
+    * ``use_azure_identity_object``: If set to true, it will use credential of newer type ClientSecretCredential or DefaultAzureCredential instead of ServicePrincipalCredentials or AzureIdentityCredentialAdapter.
+      These newer credentails support get_token method which can be used to pull OAuth token with custom scope.
 
     The entire extra column can be left out to fall back on DefaultAzureCredential_.
 
diff --git a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
index 5705dcbebed67..2e51767b6b8fb 100644
--- a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
+++ b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
@@ -134,8 +134,8 @@ def get_credential(
         """
         Get Azure credential object for the connection.
 
-        Azure Identity based credential object can be used to get token.
-        Older Credential object are supported for backward compatibility.
+        Azure Identity based credential object (ClientSecretCredential, DefaultAzureCredential) can be used to get OAuth token using `get_token` method.
+        Older Credential objects (ServicePrincipalCredentials, AzureIdentityCredentialAdapter) are supported for backward compatibility.
 
         :return: The Azure credential object
         """

From 5807c86077f503cea9d533932c055110cbbee062 Mon Sep 17 00:00:00 2001
From: karunpoudel <62040859+karunpoudel@users.noreply.github.com>
Date: Tue, 24 Jun 2025 16:02:33 -0400
Subject: [PATCH 04/18] fx

---
 .../microsoft/azure/hooks/base_azure.py         | 17 +++++------------
 1 file changed, 5 insertions(+), 12 deletions(-)

diff --git a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
index 5705dcbebed67..ee047b1c3eebe 100644
--- a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
+++ b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
@@ -16,7 +16,7 @@
 # under the License.
 from __future__ import annotations
 
-from typing import Any, Optional
+from typing import Any
 
 from azure.common.client_factory import get_client_from_auth_file, get_client_from_json_dict
 from azure.common.credentials import ServicePrincipalCredentials
@@ -24,13 +24,15 @@
 
 from airflow.exceptions import AirflowException
 from airflow.hooks.base import BaseHook
-from airflow.models import Connection
 from airflow.providers.microsoft.azure.utils import (
     AzureIdentityCredentialAdapter,
     add_managed_identity_connection_widgets,
     get_sync_default_azure_credential,
 )
 
+if TYPE_CHECKING:
+    from airflow.models import Connection
+
 
 class AzureBaseHook(BaseHook):
     """
@@ -102,7 +104,6 @@ def get_conn(self) -> Any:
         if not self.sdk_client:
             raise ValueError("`sdk_client` must be provided to AzureBaseHook to use `get_conn` method.")
         conn = self.get_connection(self.conn_id)
-        tenant = conn.extra_dejson.get("tenantId")
         subscription_id = conn.extra_dejson.get("subscriptionId")
         key_path = conn.extra_dejson.get("key_path")
         if key_path:
@@ -124,7 +125,7 @@ def get_conn(self) -> Any:
         )
 
     def get_credential(
-        self, *, conn: Optional[Connection] = None
+        self, *, conn: Connection | None = None
     ) -> (
         ServicePrincipalCredentials
         | AzureIdentityCredentialAdapter
@@ -142,14 +143,6 @@ def get_credential(
         if not conn:
             conn = self.get_connection(self.conn_id)
         tenant = conn.extra_dejson.get("tenantId")
-        if not tenant and conn.extra_dejson.get("extra__azure__tenantId"):
-            warnings.warn(
-                "`extra__azure__tenantId` is deprecated in azure connection extra, "
-                "please use `tenantId` instead",
-                AirflowProviderDeprecationWarning,
-                stacklevel=2,
-            )
-            tenant = conn.extra_dejson.get("extra__azure__tenantId")
         use_azure_identity_object = conn.extra_dejson.get("use_azure_identity_object", False)
         credential: (
             ServicePrincipalCredentials

From be69a631f8286add0d521ad118c7ccfa3a7eb45d Mon Sep 17 00:00:00 2001
From: Karun Poudel <karun.poudel@chrobinson.com>
Date: Mon, 30 Jun 2025 14:53:00 +0000
Subject: [PATCH 05/18] doc update

---
 providers/microsoft/azure/docs/connections/azure.rst | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/providers/microsoft/azure/docs/connections/azure.rst b/providers/microsoft/azure/docs/connections/azure.rst
index 79763c9f4cee0..41852363442be 100644
--- a/providers/microsoft/azure/docs/connections/azure.rst
+++ b/providers/microsoft/azure/docs/connections/azure.rst
@@ -75,7 +75,7 @@ Extra (optional)
     * ``managed_identity_client_id``:  The client ID of a user-assigned managed identity. If provided with ``workload_identity_tenant_id``, they'll pass to DefaultAzureCredential_.
     * ``workload_identity_tenant_id``: ID of the application's Microsoft Entra tenant. Also called its "directory" ID. If provided with ``managed_identity_client_id``, they'll pass to DefaultAzureCredential_.
     * ``use_azure_identity_object``: If set to true, it will use credential of newer type ClientSecretCredential or DefaultAzureCredential instead of ServicePrincipalCredentials or AzureIdentityCredentialAdapter.
-      These newer credentails support get_token method which can be used to pull OAuth token with custom scope.
+      These newer credentails support get_token method which can be used to generate OAuth token with custom scope.
 
     The entire extra column can be left out to fall back on DefaultAzureCredential_.
 

From f7ffb1532771e9373478016741f4fb96a20d19a5 Mon Sep 17 00:00:00 2001
From: Karun Poudel <64540927+karunpoudel-chr@users.noreply.github.com>
Date: Mon, 30 Jun 2025 13:41:35 -0500
Subject: [PATCH 06/18] Update azure.rst

---
 providers/microsoft/azure/docs/connections/azure.rst | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/providers/microsoft/azure/docs/connections/azure.rst b/providers/microsoft/azure/docs/connections/azure.rst
index 41852363442be..e519cd92b1eb5 100644
--- a/providers/microsoft/azure/docs/connections/azure.rst
+++ b/providers/microsoft/azure/docs/connections/azure.rst
@@ -74,7 +74,7 @@ Extra (optional)
       It specifies the json that contains the authentication information.
     * ``managed_identity_client_id``:  The client ID of a user-assigned managed identity. If provided with ``workload_identity_tenant_id``, they'll pass to DefaultAzureCredential_.
     * ``workload_identity_tenant_id``: ID of the application's Microsoft Entra tenant. Also called its "directory" ID. If provided with ``managed_identity_client_id``, they'll pass to DefaultAzureCredential_.
-    * ``use_azure_identity_object``: If set to true, it will use credential of newer type ClientSecretCredential or DefaultAzureCredential instead of ServicePrincipalCredentials or AzureIdentityCredentialAdapter.
+    * ``use_azure_identity_object``: If set to true, it will use credential of newer type: ClientSecretCredential or DefaultAzureCredential instead of ServicePrincipalCredentials or AzureIdentityCredentialAdapter.
       These newer credentails support get_token method which can be used to generate OAuth token with custom scope.
 
     The entire extra column can be left out to fall back on DefaultAzureCredential_.

From f67cd27396fb30b6f90acc502330b85d774b6073 Mon Sep 17 00:00:00 2001
From: karunpoudel <62040859+karunpoudel@users.noreply.github.com>
Date: Thu, 10 Jul 2025 10:59:53 -0400
Subject: [PATCH 07/18] fix

---
 providers/microsoft/azure/docs/connections/azure.rst            | 2 +-
 .../src/airflow/providers/microsoft/azure/hooks/base_azure.py   | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/providers/microsoft/azure/docs/connections/azure.rst b/providers/microsoft/azure/docs/connections/azure.rst
index e519cd92b1eb5..abe425067eb84 100644
--- a/providers/microsoft/azure/docs/connections/azure.rst
+++ b/providers/microsoft/azure/docs/connections/azure.rst
@@ -75,7 +75,7 @@ Extra (optional)
     * ``managed_identity_client_id``:  The client ID of a user-assigned managed identity. If provided with ``workload_identity_tenant_id``, they'll pass to DefaultAzureCredential_.
     * ``workload_identity_tenant_id``: ID of the application's Microsoft Entra tenant. Also called its "directory" ID. If provided with ``managed_identity_client_id``, they'll pass to DefaultAzureCredential_.
     * ``use_azure_identity_object``: If set to true, it will use credential of newer type: ClientSecretCredential or DefaultAzureCredential instead of ServicePrincipalCredentials or AzureIdentityCredentialAdapter.
-      These newer credentails support get_token method which can be used to generate OAuth token with custom scope.
+      These newer credentials support get_token method which can be used to generate OAuth token with custom scope.
 
     The entire extra column can be left out to fall back on DefaultAzureCredential_.
 
diff --git a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
index 94460fc366d4a..ae910733ac60f 100644
--- a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
+++ b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
@@ -16,7 +16,7 @@
 # under the License.
 from __future__ import annotations
 
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from azure.common.client_factory import get_client_from_auth_file, get_client_from_json_dict
 from azure.common.credentials import ServicePrincipalCredentials

From 957510305a0202c7b12df0ffea506162c47cb3b4 Mon Sep 17 00:00:00 2001
From: karunpoudel <62040859+karunpoudel@users.noreply.github.com>
Date: Thu, 10 Jul 2025 11:47:57 -0400
Subject: [PATCH 08/18] fix

---
 .../src/airflow/providers/microsoft/azure/hooks/base_azure.py  | 2 +-
 .../src/airflow/providers/microsoft/azure/version_compat.py    | 3 ++-
 2 files changed, 3 insertions(+), 2 deletions(-)

diff --git a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
index ae910733ac60f..6e6210f61f653 100644
--- a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
+++ b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
@@ -31,7 +31,7 @@
 from airflow.providers.microsoft.azure.version_compat import BaseHook
 
 if TYPE_CHECKING:
-    from airflow.models import Connection
+    from airflow.providers.microsoft.azure.version_compat import Connection
 
 
 class AzureBaseHook(BaseHook):
diff --git a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/version_compat.py b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/version_compat.py
index 3825100866897..85df7b4e1b1da 100644
--- a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/version_compat.py
+++ b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/version_compat.py
@@ -45,9 +45,10 @@ def get_base_airflow_version_tuple() -> tuple[int, int, int]:
         BaseOperator,
         BaseOperatorLink,
         BaseSensorOperator,
+        Connection,
     )
 else:
-    from airflow.models import BaseOperator, BaseOperatorLink  # type: ignore[no-redef]
+    from airflow.models import BaseOperator, BaseOperatorLink, Connection  # type: ignore[no-redef]
     from airflow.sensors.base import BaseSensorOperator  # type: ignore[no-redef]
 
 __all__ = [

From 4e57ceb87a38a0a2b00c70f80f658239d6db1b47 Mon Sep 17 00:00:00 2001
From: karunpoudel <62040859+karunpoudel@users.noreply.github.com>
Date: Thu, 10 Jul 2025 13:08:56 -0400
Subject: [PATCH 09/18] fx

---
 .../src/airflow/providers/microsoft/azure/version_compat.py      | 1 +
 1 file changed, 1 insertion(+)

diff --git a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/version_compat.py b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/version_compat.py
index 85df7b4e1b1da..b31ee33cdbd19 100644
--- a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/version_compat.py
+++ b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/version_compat.py
@@ -57,4 +57,5 @@ def get_base_airflow_version_tuple() -> tuple[int, int, int]:
     "BaseOperator",
     "BaseOperatorLink",
     "BaseSensorOperator",
+    "Connection",
 ]

From b9bfd18d19fc29cc1bf3871c8da3be12f5baa06a Mon Sep 17 00:00:00 2001
From: karunpoudel <62040859+karunpoudel@users.noreply.github.com>
Date: Thu, 10 Jul 2025 17:18:39 -0400
Subject: [PATCH 10/18] fix import

---
 .../src/airflow/providers/microsoft/azure/hooks/base_azure.py   | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
index 6e6210f61f653..87ca04832781e 100644
--- a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
+++ b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
@@ -31,7 +31,7 @@
 from airflow.providers.microsoft.azure.version_compat import BaseHook
 
 if TYPE_CHECKING:
-    from airflow.providers.microsoft.azure.version_compat import Connection
+    from airflow.sdk import Connection
 
 
 class AzureBaseHook(BaseHook):

From 1da12bbfc1fc2779440491e4a51bef59cd090ce9 Mon Sep 17 00:00:00 2001
From: karunpoudel <62040859+karunpoudel@users.noreply.github.com>
Date: Thu, 10 Jul 2025 17:44:46 -0400
Subject: [PATCH 11/18] fix import

---
 .../src/airflow/providers/microsoft/azure/version_compat.py    | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/version_compat.py b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/version_compat.py
index b31ee33cdbd19..8bca51ef22b15 100644
--- a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/version_compat.py
+++ b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/version_compat.py
@@ -45,10 +45,9 @@ def get_base_airflow_version_tuple() -> tuple[int, int, int]:
         BaseOperator,
         BaseOperatorLink,
         BaseSensorOperator,
-        Connection,
     )
 else:
-    from airflow.models import BaseOperator, BaseOperatorLink, Connection  # type: ignore[no-redef]
+    from airflow.models import BaseOperator, BaseOperatorLink  # type: ignore[no-redef]
     from airflow.sensors.base import BaseSensorOperator  # type: ignore[no-redef]
 
 __all__ = [

From 649944a62eeec3f378ca388e1664e3c332135117 Mon Sep 17 00:00:00 2001
From: karunpoudel <62040859+karunpoudel@users.noreply.github.com>
Date: Fri, 11 Jul 2025 16:41:25 -0400
Subject: [PATCH 12/18] fix import

---
 .../src/airflow/providers/microsoft/azure/version_compat.py      | 1 -
 1 file changed, 1 deletion(-)

diff --git a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/version_compat.py b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/version_compat.py
index 8bca51ef22b15..3825100866897 100644
--- a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/version_compat.py
+++ b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/version_compat.py
@@ -56,5 +56,4 @@ def get_base_airflow_version_tuple() -> tuple[int, int, int]:
     "BaseOperator",
     "BaseOperatorLink",
     "BaseSensorOperator",
-    "Connection",
 ]

From 2965ccfe6882ac8d930c8a0eac2f640950c8a892 Mon Sep 17 00:00:00 2001
From: Karun Poudel <64540927+karunpoudel-chr@users.noreply.github.com>
Date: Wed, 16 Jul 2025 11:42:39 -0500
Subject: [PATCH 13/18] Update base_azure.py

---
 .../airflow/providers/microsoft/azure/hooks/base_azure.py   | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
index 87ca04832781e..26f38cc9f9b4a 100644
--- a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
+++ b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
@@ -135,8 +135,8 @@ def get_credential(
         """
         Get Azure credential object for the connection.
 
-        Azure Identity based credential object (ClientSecretCredential, DefaultAzureCredential) can be used to get OAuth token using `get_token` method.
-        Older Credential objects (ServicePrincipalCredentials, AzureIdentityCredentialAdapter) are supported for backward compatibility.
+        Azure Identity based credential object (`ClientSecretCredential`, `DefaultAzureCredential`) can be used to get OAuth token using `get_token` method.
+        Older Credential objects (`ServicePrincipalCredentials`, `AzureIdentityCredentialAdapter`) are supported for backward compatibility.
 
         :return: The Azure credential object
         """
@@ -154,7 +154,7 @@ def get_credential(
             self.log.info("Getting credentials using specific credentials and subscription_id.")
             if use_azure_identity_object:
                 credential = ClientSecretCredential(
-                    client_id=conn.login, client_secret=conn.password, tenant_id=tenant
+                    client_id=conn.login, client_secret=conn.password, tenant_id=tenant  # type: ignore[arg-type]
                 )
             else:
                 credential = ServicePrincipalCredentials(

From 74aca20f5b7e0e7ef0edf938eea04e3c0d071b96 Mon Sep 17 00:00:00 2001
From: Karun Poudel <karun.poudel@chrobinson.com>
Date: Mon, 21 Jul 2025 04:29:13 +0000
Subject: [PATCH 14/18] precommit fixes

---
 .../airflow/providers/microsoft/azure/hooks/base_azure.py | 8 +++++---
 1 file changed, 5 insertions(+), 3 deletions(-)

diff --git a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
index 26f38cc9f9b4a..a3ca7fef1fbd5 100644
--- a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
+++ b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
@@ -135,8 +135,8 @@ def get_credential(
         """
         Get Azure credential object for the connection.
 
-        Azure Identity based credential object (`ClientSecretCredential`, `DefaultAzureCredential`) can be used to get OAuth token using `get_token` method.
-        Older Credential objects (`ServicePrincipalCredentials`, `AzureIdentityCredentialAdapter`) are supported for backward compatibility.
+        Azure Identity based credential object (``ClientSecretCredential``, ``DefaultAzureCredential``) can be used to get OAuth token using ``get_token`` method.
+        Older Credential objects (``ServicePrincipalCredentials``, ``AzureIdentityCredentialAdapter``) are supported for backward compatibility.
 
         :return: The Azure credential object
         """
@@ -154,7 +154,9 @@ def get_credential(
             self.log.info("Getting credentials using specific credentials and subscription_id.")
             if use_azure_identity_object:
                 credential = ClientSecretCredential(
-                    client_id=conn.login, client_secret=conn.password, tenant_id=tenant  # type: ignore[arg-type]
+                    client_id=conn.login,
+                    client_secret=conn.password,
+                    tenant_id=tenant,  # type: ignore[arg-type]
                 )
             else:
                 credential = ServicePrincipalCredentials(

From 671d875639a2d6da6cdb94c5c67e668fb7996d15 Mon Sep 17 00:00:00 2001
From: Karun Poudel <karun.poudel@chrobinson.com>
Date: Wed, 23 Jul 2025 18:12:50 +0000
Subject: [PATCH 15/18] pre-commit and doc fix

---
 .../providers/microsoft/azure/hooks/base_azure.py   | 13 +++----------
 1 file changed, 3 insertions(+), 10 deletions(-)

diff --git a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
index a3ca7fef1fbd5..39d7520751e3e 100644
--- a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
+++ b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
@@ -124,14 +124,7 @@ def get_conn(self) -> Any:
             subscription_id=subscription_id,
         )
 
-    def get_credential(
-        self, *, conn: Connection | None = None
-    ) -> (
-        ServicePrincipalCredentials
-        | AzureIdentityCredentialAdapter
-        | ClientSecretCredential
-        | DefaultAzureCredential
-    ):
+    def get_credential(self, *, conn: Connection | None = None) -> Any:
         """
         Get Azure credential object for the connection.
 
@@ -154,8 +147,8 @@ def get_credential(
             self.log.info("Getting credentials using specific credentials and subscription_id.")
             if use_azure_identity_object:
                 credential = ClientSecretCredential(
-                    client_id=conn.login,
-                    client_secret=conn.password,
+                    client_id=conn.login,  # type: ignore[arg-type]
+                    client_secret=conn.password,  # type: ignore[arg-type]
                     tenant_id=tenant,  # type: ignore[arg-type]
                 )
             else:

From 2e213d5e4fb952077633c9d1c14b95dbf7e94ec8 Mon Sep 17 00:00:00 2001
From: Karun Poudel <karun.poudel@chrobinson.com>
Date: Sun, 10 Aug 2025 05:43:07 +0000
Subject: [PATCH 16/18] refactor

---
 .../microsoft/azure/hooks/base_azure.py       | 56 ++++++++++---------
 1 file changed, 31 insertions(+), 25 deletions(-)

diff --git a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
index 39d7520751e3e..b1950c70bd281 100644
--- a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
+++ b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
@@ -136,7 +136,6 @@ def get_credential(self, *, conn: Connection | None = None) -> Any:
         if not conn:
             conn = self.get_connection(self.conn_id)
         tenant = conn.extra_dejson.get("tenantId")
-        use_azure_identity_object = conn.extra_dejson.get("use_azure_identity_object", False)
         credential: (
             ServicePrincipalCredentials
             | AzureIdentityCredentialAdapter
@@ -144,29 +143,36 @@ def get_credential(self, *, conn: Connection | None = None) -> Any:
             | DefaultAzureCredential
         )
         if all([conn.login, conn.password, tenant]):
-            self.log.info("Getting credentials using specific credentials and subscription_id.")
-            if use_azure_identity_object:
-                credential = ClientSecretCredential(
-                    client_id=conn.login,  # type: ignore[arg-type]
-                    client_secret=conn.password,  # type: ignore[arg-type]
-                    tenant_id=tenant,  # type: ignore[arg-type]
-                )
-            else:
-                credential = ServicePrincipalCredentials(
-                    client_id=conn.login, secret=conn.password, tenant=tenant
-                )
+            credential = self._get_client_secret_credential(conn)
         else:
-            self.log.info("Using DefaultAzureCredential as credential")
-            managed_identity_client_id = conn.extra_dejson.get("managed_identity_client_id")
-            workload_identity_tenant_id = conn.extra_dejson.get("workload_identity_tenant_id")
-            if use_azure_identity_object:
-                credential = get_sync_default_azure_credential(
-                    managed_identity_client_id=managed_identity_client_id,
-                    workload_identity_tenant_id=workload_identity_tenant_id,
-                )
-            else:
-                credential = AzureIdentityCredentialAdapter(
-                    managed_identity_client_id=managed_identity_client_id,
-                    workload_identity_tenant_id=workload_identity_tenant_id,
-                )
+            credential = self._get_default_azure_credential(conn)
         return credential
+
+    def _get_client_secret_credential(self, conn: Connection):
+        self.log.info("Getting credentials using specific credentials and subscription_id.")
+        extra_dejson = conn.extra_dejson
+        tenant = extra_dejson.get("tenantId")
+        use_azure_identity_object = extra_dejson.get("use_azure_identity_object", False)
+        if use_azure_identity_object:
+            return ClientSecretCredential(
+                client_id=conn.login,  # type: ignore[arg-type]
+                client_secret=conn.password,  # type: ignore[arg-type]
+                tenant_id=tenant,  # type: ignore[arg-type]
+            )
+        return ServicePrincipalCredentials(client_id=conn.login, secret=conn.password, tenant=tenant)
+
+    def _get_default_azure_credential(self, conn: Connection):
+        self.log.info("Using DefaultAzureCredential as credential")
+        extra_dejson = conn.extra_dejson
+        managed_identity_client_id = extra_dejson.get("managed_identity_client_id")
+        workload_identity_tenant_id = extra_dejson.get("workload_identity_tenant_id")
+        use_azure_identity_object = extra_dejson.get("use_azure_identity_object", False)
+        if use_azure_identity_object:
+            return get_sync_default_azure_credential(
+                managed_identity_client_id=managed_identity_client_id,
+                workload_identity_tenant_id=workload_identity_tenant_id,
+            )
+        return AzureIdentityCredentialAdapter(
+            managed_identity_client_id=managed_identity_client_id,
+            workload_identity_tenant_id=workload_identity_tenant_id,
+        )

From df5d7dcec0a31960056cc08c377a5e4352bf767d Mon Sep 17 00:00:00 2001
From: Karun Poudel <karun.poudel@chrobinson.com>
Date: Sun, 10 Aug 2025 16:02:58 +0000
Subject: [PATCH 17/18] included get_token method in hook for error handling

---
 .../providers/microsoft/azure/hooks/base_azure.py  | 14 ++++++++++++++
 1 file changed, 14 insertions(+)

diff --git a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
index b1950c70bd281..5693321676088 100644
--- a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
+++ b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/base_azure.py
@@ -31,6 +31,8 @@
 from airflow.providers.microsoft.azure.version_compat import BaseHook
 
 if TYPE_CHECKING:
+    from azure.core.credentials import AccessToken
+
     from airflow.sdk import Connection
 
 
@@ -176,3 +178,15 @@ def _get_default_azure_credential(self, conn: Connection):
             managed_identity_client_id=managed_identity_client_id,
             workload_identity_tenant_id=workload_identity_tenant_id,
         )
+
+    def get_token(self, *scopes, **kwargs) -> AccessToken:
+        """Request an access token for `scopes`."""
+        credential = self.get_credential()
+        if isinstance(credential, AzureIdentityCredentialAdapter) or isinstance(
+            credential, AzureIdentityCredentialAdapter
+        ):
+            raise AttributeError(
+                "The azure credential does not support get_token method. "
+                "Please set `use_azure_identity_object: True` in the connection extra field to use credential that support get_token method."
+            )
+        return credential.get_token(*scopes, **kwargs)

From 69d906c08eaa72124b2d77405fb0a8a938443f74 Mon Sep 17 00:00:00 2001
From: Karun Poudel <karun.poudel@chrobinson.com>
Date: Sun, 10 Aug 2025 16:26:25 +0000
Subject: [PATCH 18/18] add test for get_token

---
 .../microsoft/azure/hooks/test_base_azure.py  | 19 +++++++++++++++++++
 1 file changed, 19 insertions(+)

diff --git a/providers/microsoft/azure/tests/unit/microsoft/azure/hooks/test_base_azure.py b/providers/microsoft/azure/tests/unit/microsoft/azure/hooks/test_base_azure.py
index aafcac670059a..98df1a4c08fcc 100644
--- a/providers/microsoft/azure/tests/unit/microsoft/azure/hooks/test_base_azure.py
+++ b/providers/microsoft/azure/tests/unit/microsoft/azure/hooks/test_base_azure.py
@@ -202,3 +202,22 @@ def test_get_credential_with_azure_default_credential_with_extra(self, mock_spc,
             additionally_allowed_tenants=[mocked_connection.extra_dejson.get("workload_identity_tenant_id")],
         )
         assert cred == "foo-bar"
+
+    @patch(f"{UTILS}.DefaultAzureCredential")
+    @pytest.mark.parametrize(
+        "mocked_connection",
+        [
+            Connection(
+                conn_id="azure_default",
+                extra={"use_azure_identity_object": True},
+            ),
+        ],
+        indirect=True,
+    )
+    def test_get_token_with_azure_default_credential(self, mock_spc, mocked_connection):
+        mock_spc.get_token.return_value = "new-token"
+        scope = "custom_scope"
+        token = AzureBaseHook().get_token(scope)
+
+        mock_spc.assert_called_once_with()
+        assert token == "new-token"