@@ -70,7 +70,7 @@ With Object Storage or a Database Table
 
 Use ``data_sources`` with
 :class:`~airflow.providers.common.sql.config.DataSourceConfig` to include
-object-storage sources (S3 or GCS Parquet, CSV, Iceberg, etc.) in the comparison.
+object-storage sources (S3, GCS, Azure Blob Storage, Parquet, CSV, Iceberg, etc.) in the comparison.
 These can be freely combined with ``db_conn_ids``. Whether an entry is
 introspected via ``DbApiHook`` or DataFusion depends on what its ``conn_id``
 resolves to, not on its ``uri``/``format`` fields: a ``DataSourceConfig``
 
@@ -65,7 +65,7 @@ With Object Storage
 -------------------
 
 Use ``datasource_config`` to generate queries for data stored in object storage
-(e.g., S3, GCS, local filesystem) via `DataFusion <https://datafusion.apache.org/>`_.
+(e.g., S3, GCS, Azure Blob Storage, local filesystem) via `DataFusion <https://datafusion.apache.org/>`_.
 The operator uses :class:`~airflow.providers.common.sql.config.DataSourceConfig`
 to register the object storage source as a table so the LLM can include it in
 the schema context.
 
@@ -26,7 +26,7 @@ Files with DataFusion: ``DataFusionToolset``
 Curated toolset wrapping
 :class:`~airflow.providers.common.sql.datafusion.engine.DataFusionEngine`
 with three tools (``list_tables``, ``get_schema``, and ``query``) for
-querying files on object stores (S3, GCS, local filesystem, Iceberg) via Apache DataFusion.
+querying files on object stores (S3, GCS, Azure Blob Storage, local filesystem, Iceberg) via Apache DataFusion.
 
 .. list-table::
    :header-rows: 1
 
@@ -297,10 +297,11 @@ Supported Storage Systems
 -------------------------
 - S3
 - GCS
+- Azure Blob Storage
 - Local File System
 
 .. note::
-   Azure, HTTP, Delta are not yet supported but will be added in the future.
+   HTTP, Delta are not yet supported but will be added in the future.
 
 
 
@@ -362,6 +363,63 @@ resolved in this order:
     :start-after: [START howto_analytics_operator_with_gcs]
     :end-before: [END howto_analytics_operator_with_gcs]
 
+Azure Storage
+-------------
+Use an ``az://`` URI with a ``conn_id`` pointing to a ``wasb`` connection.
+``abfs://`` and ``abfss://`` URIs are not recognized yet. The account name
+comes from ``host`` (its first DNS label) when set, falling back to
+``login`` only when ``host`` is empty; only the public
+``*.blob.core.windows.net`` cloud is supported, since DataFusion's binding
+otherwise has no endpoint override -- unless the worker sets
+``AZURE_STORAGE_ENDPOINT``/``AZURE_ENDPOINT`` for a real sovereign-cloud
+hostname. A ``host:port`` address (the Azurite emulator's shape, e.g.
+``azurite:10000``) is never a real DNS hostname and always raises instead,
+even with those variables set: put the account name in ``login`` with
+``host`` left empty for Azurite, alongside those same variables and
+``AZURE_ALLOW_HTTP=true``. ``client_secret_auth_config`` (the authority
+override ``WasbHook`` honors) is not read here.
+
+The connection supplies one of the following credentials, checked in this
+order (matching ``WasbHook.get_conn``):
+
+1. Azure AD service principal -- ``tenant_id`` extra, with ``login`` as the
+   client ID and ``password`` as the client secret (both required together)
+2. Shared key -- the ``shared_access_key`` extra
+3. SAS token -- ``sas_token`` extra, as a query string
+4. Shared key -- ``password``, or the ``account_key`` extra
+5. None of the above -- ambient auth (see below)
+
+**A worker environment variable can override the connection, but only from a
+higher-priority tier.** DataFusion reads ``AZURE_*`` environment variables
+first, and resolves credentials in this priority order:
+
+1. Bearer token
+2. Access key -- the shared-key connection's tier
+3. Workload identity
+4. Client secret -- the service-principal connection's tier
+5. SAS -- the SAS connection's tier
+
+Only a complete credential from an earlier tier can override the
+connection -- a single bearer or access-key variable, or every variable a
+multi-field tier needs. This raises when found, naming the variables,
+instead of silently using the wrong identity.
+
+With no explicit credential, authentication falls back to ``AZURE_*``
+environment variables, managed identity, or workload identity. Unlike
+``WasbHook`` (which tries ``az login`` automatically), DataFusion's
+underlying ``object_store`` binding only tries the Azure CLI if
+``AZURE_USE_AZURE_CLI=true`` is set; otherwise it defaults straight to
+IMDS managed identity.
+
+``connection_string``, ``managed_identity_client_id``, ``workload_identity_tenant_id``,
+and a URL-form ``sas_token`` are not supported.
+
+.. exampleinclude:: /../../sql/src/airflow/providers/common/sql/example_dags/example_analytics.py
+    :language: python
+    :dedent: 4
+    :start-after: [START howto_analytics_operator_with_azure]
+    :end-before: [END howto_analytics_operator_with_azure]
+
 Local File System Storage
 -------------------------
 .. exampleinclude:: /../../sql/src/airflow/providers/common/sql/example_dags/example_analytics.py
 
@@ -48,6 +48,7 @@ class StorageType(str, Enum):
 
     S3 = "s3"
     GCS = "gcs"
+    AZURE = "azure"
     LOCAL = "local"
 
 
@@ -117,6 +118,8 @@ def _extract_storage_type(self) -> StorageType | None:
             return StorageType.S3
         if self.uri.startswith("gs://"):
             return StorageType.GCS
+        if self.uri.startswith("az://"):
+            return StorageType.AZURE
         if self.uri.startswith("file://"):
             return StorageType.LOCAL
         raise ValueError(f"Unsupported storage type for URI: {self.uri}")
@@ -18,6 +18,7 @@
 
 import os
 from typing import TYPE_CHECKING, Any
+from urllib.parse import parse_qsl, urlsplit
 
 from datafusion import SessionContext
 
@@ -161,6 +162,12 @@ def _get_gcp_extra_field(extra_dejson: dict[str, Any], field_name: str) -> Any:
                 return extra_dejson[field_name]
             return extra_dejson.get(f"extra__google_cloud_platform__{field_name}")
 
+        def _get_wasb_extra_field(extra_dejson: dict[str, Any], field_name: str) -> Any:
+            # Older Airflow connection UIs wrote custom extra fields as
+            # extra__wasb__<field_name> instead of the bare key; WasbHook still reads that
+            # legacy spelling as a fallback, so this must too.
+            return extra_dejson.get(field_name, extra_dejson.get(f"extra__wasb__{field_name}"))
+
         match conn.conn_type:
             case "aws":
                 try:
@@ -204,6 +211,66 @@ def _get_gcp_extra_field(extra_dejson: dict[str, Any], field_name: str) -> Any:
                     key_path = os.environ.get("GOOGLE_APPLICATION_CREDENTIALS")
                 credentials = self._remove_none_values({"key_path": key_path, "keyfile_dict": keyfile_dict})
 
+            case "wasb":
+                extra_dejson = conn.extra_dejson
+                for unsupported_field in (
+                    "connection_string",
+                    "managed_identity_client_id",
+                    "workload_identity_tenant_id",
+                ):
+                    if _get_wasb_extra_field(extra_dejson, unsupported_field):
+                        raise ValueError(
+                            f"Connection field {unsupported_field!r} is not supported for DataFusion "
+                            "Azure Blob Storage access; only tenant_id+login+password (service "
+                            "principal), sas_token, shared_access_key/account_key/password, or ambient "
+                            "credentials (AZURE_* environment variables, managed identity, workload "
+                            "identity, or az login) are used."
+                        )
+                credentials = {"account": self._resolve_wasb_account(conn.host, conn.login)}
+                credential_tier: str | None = None
+                if tenant_id := _get_wasb_extra_field(extra_dejson, "tenant_id"):
+                    if not conn.login or not conn.password:
+                        # Falling through here would silently switch identity (ambient auth, or
+                        # the client secret sent as a shared key) instead of failing clearly.
+                        missing = "login (client_id)" if not conn.login else "password (client_secret)"
+                        raise ValueError(
+                            f"Connection extra 'tenant_id' is set for DataFusion Azure Blob Storage "
+                            f"service-principal auth, but {missing} is not."
+                        )
+                    credentials.update(
+                        {"client_id": conn.login, "client_secret": conn.password, "tenant_id": tenant_id}
+                    )
+                    credential_tier = "client_secret"
+                elif shared_access_key := _get_wasb_extra_field(extra_dejson, "shared_access_key"):
+                    # Checked ahead of sas_token to match WasbHook.get_conn's precedence.
+                    credentials["access_key"] = shared_access_key
+                    credential_tier = "access_key"
+                elif sas_token := _get_wasb_extra_field(extra_dejson, "sas_token"):
+                    if sas_token.startswith("http"):
+                        raise ValueError(
+                            "A URL-form `sas_token` is not supported for DataFusion Azure Blob Storage "
+                            "access; provide the SAS token as a query string instead."
+                        )
+                    credentials["sas_query_pairs"] = parse_qsl(sas_token.lstrip("?"))
+                    credential_tier = "sas"
+                else:
+                    access_key = conn.password or _get_wasb_extra_field(extra_dejson, "account_key")
+                    if access_key:
+                        credentials["access_key"] = access_key
+                        credential_tier = "access_key"
+
+                if credential_tier is not None:
+                    conflicting_env_vars = self._find_conflicting_azure_env_vars(credential_tier)
+                    if conflicting_env_vars:
+                        raise ValueError(
+                            f"Worker environment variable(s) {', '.join(conflicting_env_vars)} would "
+                            "silently take precedence over this connection's explicit credential in "
+                            "DataFusion's Azure Blob Storage binding. Unset them on the worker, or "
+                            "remove the explicit credential from this connection to rely on the "
+                            "environment instead."
+                        )
+                credentials = self._remove_none_values(credentials)
+
             case _:
                 raise ValueError(f"Unknown connection type {conn.conn_type}")
         return credentials, extra_config
@@ -213,6 +280,124 @@ def _remove_none_values(params: dict[str, Any]) -> dict[str, Any]:
         """Filter out None values from the dictionary."""
         return {k: v for k, v in params.items() if v is not None}
 
+    _AZURE_PUBLIC_SUFFIX = ".blob.core.windows.net"
+
+    # object_store's build() precedence, high to low: bearer token > access key > workload
+    # identity (client_id+tenant_id+federated_token_file) > client secret
+    # (client_id+client_secret+tenant_id) > SAS. Used by _find_conflicting_azure_env_vars.
+    _AZURE_ENV_BEARER_VARS = ("AZURE_STORAGE_TOKEN",)
+    _AZURE_ENV_ACCESS_KEY_VARS = (
+        "AZURE_STORAGE_ACCOUNT_KEY",
+        "AZURE_STORAGE_ACCESS_KEY",
+        "AZURE_STORAGE_MASTER_KEY",
+    )
+    _AZURE_ENV_CLIENT_ID_VARS = ("AZURE_STORAGE_CLIENT_ID", "AZURE_CLIENT_ID")
+    _AZURE_ENV_CLIENT_SECRET_VARS = ("AZURE_STORAGE_CLIENT_SECRET", "AZURE_CLIENT_SECRET")
+    _AZURE_ENV_TENANT_ID_VARS = (
+        "AZURE_STORAGE_TENANT_ID",
+        "AZURE_STORAGE_AUTHORITY_ID",
+        "AZURE_TENANT_ID",
+        "AZURE_AUTHORITY_ID",
+    )
+    _AZURE_ENV_FEDERATED_TOKEN_FILE_VAR = "AZURE_FEDERATED_TOKEN_FILE"
+
+    @classmethod
+    def _find_conflicting_azure_env_vars(cls, credential_tier: str) -> list[str]:
+        """
+        Return worker env vars that would silently outrank the connection's own credential.
+
+        The binding always calls `from_env()` with no way to skip it, so this can only be
+        caught here, not avoided. The connection's fields overwrite the same fields from
+        `from_env()`, so only a tier above the connection's own, or an unoccupied tier
+        (workload identity or client secret, for SAS) with *all* its fields present, can
+        actually take over.
+        """
+
+        def env_set(*var_groups: tuple[str, ...]) -> list[str]:
+            return [var for group in var_groups for var in group if os.environ.get(var)]
+
+        conflicting = env_set(cls._AZURE_ENV_BEARER_VARS)
+        if credential_tier == "access_key":
+            return conflicting
+        conflicting += env_set(cls._AZURE_ENV_ACCESS_KEY_VARS)
+        if credential_tier == "client_secret":
+            # client_id and tenant_id are already the connection's own; only the federated
+            # token file is left for env to complete the workload-identity triple with.
+            if os.environ.get(cls._AZURE_ENV_FEDERATED_TOKEN_FILE_VAR):
+                conflicting.append(cls._AZURE_ENV_FEDERATED_TOKEN_FILE_VAR)
+            return conflicting
+        # SAS occupies none of these fields, so each whole triple must come from env.
+        if (
+            any(os.environ.get(var) for var in cls._AZURE_ENV_CLIENT_ID_VARS)
+            and any(os.environ.get(var) for var in cls._AZURE_ENV_TENANT_ID_VARS)
+            and os.environ.get(cls._AZURE_ENV_FEDERATED_TOKEN_FILE_VAR)
+        ):
+            conflicting += env_set(cls._AZURE_ENV_CLIENT_ID_VARS, cls._AZURE_ENV_TENANT_ID_VARS)
+            conflicting.append(cls._AZURE_ENV_FEDERATED_TOKEN_FILE_VAR)
+        if (
+            any(os.environ.get(var) for var in cls._AZURE_ENV_CLIENT_ID_VARS)
+            and any(os.environ.get(var) for var in cls._AZURE_ENV_CLIENT_SECRET_VARS)
+            and any(os.environ.get(var) for var in cls._AZURE_ENV_TENANT_ID_VARS)
+        ):
+            conflicting += env_set(
+                cls._AZURE_ENV_CLIENT_ID_VARS,
+                cls._AZURE_ENV_CLIENT_SECRET_VARS,
+                cls._AZURE_ENV_TENANT_ID_VARS,
+            )
+        return list(dict.fromkeys(conflicting))
+
+    @classmethod
+    def _resolve_wasb_account(cls, host: str | None, login: str | None) -> str | None:
+        """
+        Return the storage account name the way WasbHook resolves it.
+
+        From ``host`` (its netloc's first label) when set, falling back to ``login`` only when
+        ``host`` is empty -- login holds the service-principal client_id otherwise, not the
+        account name. Returns ``None`` when neither is set, so the binding falls back to
+        ``AZURE_STORAGE_ACCOUNT_NAME`` rather than the literal string ``"None"``. Reimplemented
+        locally instead of importing
+        ``airflow.providers.microsoft.azure.utils.parse_blob_account_url``, to avoid pulling in
+        the microsoft-azure provider's Azure SDK dependency for one stdlib string operation.
+
+        Only the public ``*.blob.core.windows.net`` cloud is supported, unless
+        ``AZURE_STORAGE_ENDPOINT``/``AZURE_ENDPOINT`` is set for a real sovereign-cloud hostname
+        (DataFusion's binding otherwise has no endpoint override). A ``host:port`` netloc (the
+        Azurite emulator's shape) is never a real hostname, so it always raises instead --
+        checked before the dotless-host fallback below, which would otherwise mask it.
+        """
+        if not host and not login:
+            return None
+        netloc = urlsplit(host if host else f"https://{login}.blob.core.windows.net/").netloc
+        if not netloc:
+            # No scheme was given (e.g. a bare DNS name); urlsplit put it all in the path instead.
+            netloc = urlsplit(f"https://{host}").netloc
+        if host and ":" in netloc:
+            raise ValueError(
+                f"Connection host {host!r} looks like an emulator address (host:port), which "
+                "DataFusion's Azure Blob Storage binding cannot resolve an account name from -- "
+                "even with AZURE_STORAGE_ENDPOINT set. Put the account name in `login` with "
+                "`host` empty instead, alongside AZURE_STORAGE_ENDPOINT and AZURE_ALLOW_HTTP=true."
+            )
+        if "." not in netloc:
+            if not login:
+                raise ValueError(
+                    f"Connection host {host!r} is not a full URL or DNS name, and no `login` was "
+                    "given to resolve it as an Active Directory ID instead."
+                )
+            # Only an Active Directory ID was given, not a full URL or DNS name.
+            netloc = f"{login}.blob.core.windows.net"
+        if not netloc.endswith(cls._AZURE_PUBLIC_SUFFIX) and not (
+            os.environ.get("AZURE_STORAGE_ENDPOINT") or os.environ.get("AZURE_ENDPOINT")
+        ):
+            raise ValueError(
+                f"Connection host {host!r} does not resolve to the public {cls._AZURE_PUBLIC_SUFFIX} "
+                "cloud, which is the only one DataFusion's Azure Blob Storage binding can target (it "
+                "has no endpoint override). A sovereign cloud is supported once the "
+                "AZURE_STORAGE_ENDPOINT environment variable is set."
+            )
+        # Azure storage account names are capped at 24 characters.
+        return netloc.split(".", 1)[0][:24]
+
     def get_schema(self, table_name: str):
         """Get the schema of a table."""
         schema = str(self.session_context.table(table_name).schema())
 
@@ -20,7 +20,7 @@
 import tempfile
 from pathlib import Path
 
-from datafusion.object_store import AmazonS3, GoogleCloud, LocalFileSystem
+from datafusion.object_store import AmazonS3, GoogleCloud, LocalFileSystem, MicrosoftAzure
 
 from airflow.providers.common.sql.config import ConnectionConfig, StorageType
 from airflow.providers.common.sql.datafusion.base import ObjectStorageProvider
@@ -107,6 +107,37 @@ def get_scheme(self) -> str:
         return "gs://"
 
 
+class AzureObjectStorageProvider(ObjectStorageProvider):
+    """Azure Object Storage Provider using DataFusion's MicrosoftAzure."""
+
+    @property
+    def get_storage_type(self) -> StorageType:
+        """Return the storage type."""
+        return StorageType.AZURE
+
+    def create_object_store(self, path: str, connection_config: ConnectionConfig | None = None):
+        """Create an Azure object store using DataFusion's MicrosoftAzure."""
+        if connection_config is None:
+            raise ValueError(f"connection_config must be provided for {self.get_storage_type.value}")
+
+        try:
+            credentials = connection_config.credentials
+            container = self.get_bucket(path)
+
+            azure_store = MicrosoftAzure(container_name=container, **credentials)
+            self.log.info("Created Azure object store for container %s", container)
+
+            return azure_store
+
+        except BaseException as e:
+            # A bad credential combination panics as pyo3_runtime.PanicException, not a plain Exception.
+            raise ObjectStoreCreationException(f"Failed to create Azure object store: {e}")
+
+    def get_scheme(self) -> str:
+        """Return the scheme for Azure."""
+        return "az://"
+
+
 class LocalObjectStorageProvider(ObjectStorageProvider):
     """Local Object Storage Provider using DataFusion's LocalFileSystem."""
 
@@ -126,10 +157,11 @@ def get_scheme(self) -> str:
 
 def get_object_storage_provider(storage_type: StorageType) -> ObjectStorageProvider:
     """Get an object storage provider based on the storage type."""
-    # TODO: Add support for Azure, HTTP: https://datafusion.apache.org/python/autoapi/datafusion/object_store/index.html
+    # TODO: Add support for HTTP: https://datafusion.apache.org/python/autoapi/datafusion/object_store/index.html
     providers: dict[StorageType, type] = {
         StorageType.S3: S3ObjectStorageProvider,
         StorageType.GCS: GCSObjectStorageProvider,
+        StorageType.AZURE: AzureObjectStorageProvider,
         StorageType.LOCAL: LocalObjectStorageProvider,
     }