Azure Message Bus - CommonMessageQueue Interface (#52712)

docs/spelling_wordlist.txt

@@ -94,6 +94,7 @@ arraysize
 artifactId
 Asana
 asana
+asb
 asc
 ascii
 asciiart

providers/microsoft/azure/docs/index.rst

@@ -35,6 +35,7 @@
     :caption: Guides
 
     Connection types <connections/index>
+    Message queues <message-queues/index>
     Operators <operators/index>
     Transfers <transfer/index>
     Filesystems <filesystems/index>
@@ -151,14 +152,15 @@ You can install such cross-provider dependencies when installing from PyPI. For
     pip install apache-airflow-providers-microsoft-azure[amazon]
 
 
-==================================================================================================================  =================
-Dependent package                                                                                                   Extra
-==================================================================================================================  =================
-`apache-airflow-providers-amazon <https://airflow.apache.org/docs/apache-airflow-providers-amazon>`_                ``amazon``
-`apache-airflow-providers-common-compat <https://airflow.apache.org/docs/apache-airflow-providers-common-compat>`_  ``common.compat``
-`apache-airflow-providers-oracle <https://airflow.apache.org/docs/apache-airflow-providers-oracle>`_                ``oracle``
-`apache-airflow-providers-sftp <https://airflow.apache.org/docs/apache-airflow-providers-sftp>`_                    ``sftp``
-==================================================================================================================  =================
+========================================================================================================================  ====================
+Dependent package                                                                                                         Extra
+========================================================================================================================  ====================
+`apache-airflow-providers-amazon <https://airflow.apache.org/docs/apache-airflow-providers-amazon>`_                      ``amazon``
+`apache-airflow-providers-common-compat <https://airflow.apache.org/docs/apache-airflow-providers-common-compat>`_        ``common.compat``
+`apache-airflow-providers-common-messaging <https://airflow.apache.org/docs/apache-airflow-providers-common-messaging>`_  ``common.messaging``
+`apache-airflow-providers-oracle <https://airflow.apache.org/docs/apache-airflow-providers-oracle>`_                      ``oracle``
+`apache-airflow-providers-sftp <https://airflow.apache.org/docs/apache-airflow-providers-sftp>`_                          ``sftp``
+========================================================================================================================  ====================
 
 Downloading official packages
 -----------------------------

providers/microsoft/azure/docs/message-queues/index.rst

@@ -0,0 +1,69 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements.  See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership.  The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License.  You may obtain a copy of the License at
+
+ ..   http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied.  See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+Azure Service Bus Message Queue
+================================
+
+.. contents::
+   :local:
+   :depth: 2
+
+Azure Service Bus Queue Provider
+---------------------------------
+
+Implemented by :class:`~airflow.providers.microsoft.azure.queues.asb.AzureServiceBusMessageQueueProvider`
+
+The Azure Service Bus Queue Provider is a :class:`~airflow.providers.common.messaging.providers.base_provider.BaseMessageQueueProvider` that uses
+Azure Service Bus as the underlying message queue system.
+It allows you to send and receive messages using Azure Service Bus queues in your Airflow workflows with :class:`~airflow.providers.common.messaging.triggers.msg_queue.MessageQueueTrigger` common message queue interface.
+
+
+.. include:: /../src/airflow/providers/microsoft/azure/queues/asb.py
+    :start-after: [START azure_servicebus_message_queue_provider_description]
+    :end-before: [END azure_servicebus_message_queue_provider_description]
+
+Azure Service Bus Message Queue Trigger
+-----------------------------------------
+
+Implemented by :class:`~airflow.providers.microsoft.azure.triggers.message_bus.AzureServiceBusQueueTrigger`
+
+Inherited from :class:`~airflow.providers.common.messaging.triggers.msg_queue.MessageQueueTrigger`
+
+Wait for a message in a queue
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Below is an example of how you can configure an Airflow DAG to be triggered by a message in Azure Service Bus.
+
+.. exampleinclude:: /../tests/system/microsoft/azure/example_event_schedule_asb.py
+    :language: python
+    :start-after: [START howto_trigger_asb_message_queue]
+    :end-before: [END howto_trigger_asb_message_queue]
+
+How it works
+------------
+
+1. **Azure Service Bus Message Queue Trigger**: The ``AzureServiceBusQueueTrigger`` listens for messages from Azure Service Bus queue(s).
+
+2. **Asset and Watcher**: The ``Asset`` abstracts the external entity, the Azure Service Bus queue in this example.
+   The ``AssetWatcher`` associate a trigger with a name. This name helps you identify which trigger is associated to which
+   asset.
+
+3. **Event-Driven DAG**: Instead of running on a fixed schedule, the DAG executes when the asset receives an update
+   (e.g., a new message in the queue).
+
+For how to use the trigger, refer to the documentation of the
+:ref:`Messaging Trigger <howto/trigger:MessageQueueTrigger>`

providers/microsoft/azure/provider.yaml

@@ -304,6 +304,9 @@ triggers:
     python-modules:
       - airflow.providers.microsoft.azure.triggers.message_bus
 
+queues:
+  - airflow.providers.microsoft.azure.queues.asb.AzureServiceBusMessageQueueProvider
+
 transfers:
   - source-integration-name: Local
     target-integration-name: Microsoft Azure Data Lake Storage

providers/microsoft/azure/pyproject.toml

@@ -107,6 +107,9 @@ dependencies = [
 "sftp" = [
     "apache-airflow-providers-sftp"
 ]
+"common.messaging" = [
+    "apache-airflow-providers-common-messaging>=2.0.0"
+]
 
 [dependency-groups]
 dev = [
@@ -115,6 +118,7 @@ dev = [
     "apache-airflow-devel-common",
     "apache-airflow-providers-amazon",
     "apache-airflow-providers-common-compat",
+    "apache-airflow-providers-common-messaging",
     "apache-airflow-providers-oracle",
     "apache-airflow-providers-sftp",
     # Additional devel dependencies (do not remove this line and add extra development dependencies)

providers/microsoft/azure/src/airflow/providers/microsoft/azure/get_provider_info.py

@@ -286,6 +286,7 @@ def get_provider_info():
                 "python-modules": ["airflow.providers.microsoft.azure.triggers.message_bus"],
             },
         ],
+        "queues": ["airflow.providers.microsoft.azure.queues.asb.AzureServiceBusMessageQueueProvider"],
         "transfers": [
             {
                 "source-integration-name": "Local",

providers/microsoft/azure/src/airflow/providers/microsoft/azure/queues/__init__.py

@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.

providers/microsoft/azure/src/airflow/providers/microsoft/azure/queues/asb.py

@@ -0,0 +1,62 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from airflow.providers.common.messaging.providers.base_provider import BaseMessageQueueProvider
+from airflow.providers.microsoft.azure.triggers.message_bus import AzureServiceBusQueueTrigger
+
+if TYPE_CHECKING:
+    from airflow.triggers.base import BaseEventTrigger
+
+
+class AzureServiceBusMessageQueueProvider(BaseMessageQueueProvider):
+    """
+    Configuration for Azure Service Bus integration with common-messaging.
+
+    [START azure_servicebus_message_queue_provider_description]
+
+    * It uses ``azure+servicebus`` as the scheme for identifying the provider.
+    * For parameter definitions, take a look at
+      :class:`~airflow.providers.microsoft.azure.triggers.message_bus.AzureServiceBusQueueTrigger`.
+
+    .. code-block:: python
+
+        from airflow.providers.common.messaging.triggers.msg_queue import MessageQueueTrigger
+        from airflow.sdk import Asset, AssetWatcher
+
+        trigger = MessageQueueTrigger(
+            scheme="azure+servicebus",
+            # AzureServiceBusQueueTrigger parameters
+            queues=["my-queue"],
+            azure_service_bus_conn_id="azure_service_bus_default",
+            poll_interval=60,
+        )
+
+        asset = Asset(
+            "asb_queue_asset",
+            watchers=[AssetWatcher(name="asb_watcher", trigger=trigger)],
+        )
+
+    [END azure_servicebus_message_queue_provider_description]
+    """
+
+    scheme = "azure+servicebus"
+
+    def trigger_class(self) -> type[BaseEventTrigger]:
+        return AzureServiceBusQueueTrigger

providers/microsoft/azure/tests/system/microsoft/azure/example_event_schedule_asb.py

@@ -0,0 +1,74 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+Example Airflow DAG that demonstrates using Azure Service Bus with MessageQueueTrigger
+and Asset Watchers for event-driven workflows.
+
+This example shows how to create a DAG that triggers when messages arrive in an
+Azure Service Bus queue using Asset Watchers.
+
+Prerequisites
+-------------
+
+Before running this example, ensure you have:
+
+1. An Azure Service Bus namespace
+2. A queue created in the namespace (e.g., ``my-queue``)
+3. An Airflow connection configured with ID ``azure_service_bus_default``
+
+How to test
+-----------
+
+1. Ensure the Azure Service Bus queue exists (see Prerequisites above)
+2. Send a message to the queue to trigger the DAG
+3. The DAG will be triggered automatically when the message arrives
+"""
+
+from __future__ import annotations
+
+# [START howto_trigger_asb_message_queue]
+from airflow.providers.common.messaging.triggers.msg_queue import MessageQueueTrigger
+from airflow.providers.standard.operators.empty import EmptyOperator
+from airflow.sdk import DAG, Asset, AssetWatcher
+
+# Define a trigger that listens to an Azure Service Bus queue
+trigger = MessageQueueTrigger(
+    scheme="azure+servicebus",
+    queues=["my-queue"],
+    azure_service_bus_conn_id="azure_service_bus_default",
+    poll_interval=60,
+)
+
+# Define an asset that watches for messages on the Azure Service Bus queue
+asset = Asset(
+    "event_schedule_asb_asset_1",
+    watchers=[AssetWatcher(name="event_schedule_asb_watcher_1", trigger=trigger)],
+)
+
+with DAG(
+    dag_id="example_event_schedule_asb",
+    schedule=[asset],
+) as dag:
+    process_message_task = EmptyOperator(task_id="process_asb_message")
+# [END howto_trigger_asb_message_queue]
+
+
+from tests_common.test_utils.system_tests import get_test_run  # noqa: E402
+
+# Needed to run the example DAG with pytest (see: tests/system/README.md#run_via_pytest)
+test_run = get_test_run(dag)

providers/microsoft/azure/tests/unit/microsoft/azure/queues/__init__.py

@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.

providers/microsoft/azure/tests/unit/microsoft/azure/queues/test_asb.py

@@ -0,0 +1,78 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from unittest.mock import patch
+
+import pytest
+
+from airflow.providers.microsoft.azure.triggers.message_bus import AzureServiceBusQueueTrigger
+
+from tests_common.test_utils.common_msg_queue import mark_common_msg_queue_test
+
+pytest.importorskip("airflow.providers.common.messaging.providers.base_provider")
+
+
+class TestAzureServiceBusMessageQueueProvider:
+    """Tests for AzureServiceBusMessageQueueProvider."""
+
+    def setup_method(self):
+        """Set up the test environment."""
+        from airflow.providers.microsoft.azure.queues.asb import AzureServiceBusMessageQueueProvider
+
+        self.provider = AzureServiceBusMessageQueueProvider()
+
+    def test_queue_create(self):
+        """Test the creation of the AzureServiceBusMessageQueueProvider."""
+        from airflow.providers.common.messaging.providers.base_provider import BaseMessageQueueProvider
+
+        assert isinstance(self.provider, BaseMessageQueueProvider)
+
+    @pytest.mark.parametrize(
+        ("scheme", "expected_result"),
+        [
+            pytest.param("azure+servicebus", True, id="azure_servicebus_scheme"),
+            pytest.param("sqs", False, id="sqs_scheme"),
+            pytest.param("kafka", False, id="kafka_scheme"),
+            pytest.param("redis+pubsub", False, id="redis_pubsub_scheme"),
+            pytest.param("google+pubsub", False, id="google_pubsub_scheme"),
+            pytest.param("azure", False, id="azure_only_scheme"),
+            pytest.param("unknown", False, id="unknown_scheme"),
+        ],
+    )
+    def test_scheme_matches(self, scheme, expected_result):
+        """Test the scheme_matches method with various schemes."""
+        assert self.provider.scheme_matches(scheme) == expected_result
+
+    def test_trigger_class(self):
+        """Test the trigger_class method."""
+        assert self.provider.trigger_class() == AzureServiceBusQueueTrigger
+
+
+@mark_common_msg_queue_test
+class TestMessageQueueTriggerIntegration:
+    """Test integration with MessageQueueTrigger framework."""
+
+    @pytest.mark.usefixtures("cleanup_providers_manager")
+    def test_provider_integrations_with_scheme_param(self):
+        """Test that MessageQueueTrigger correctly resolves to AzureServiceBusQueueTrigger."""
+        from airflow.providers.common.messaging.triggers.msg_queue import MessageQueueTrigger
+        from airflow.providers.microsoft.azure.triggers.message_bus import AzureServiceBusQueueTrigger
+
+        with patch("airflow.providers.microsoft.azure.triggers.message_bus.MessageHook"):
+            trigger = MessageQueueTrigger(scheme="azure+servicebus", queues=["test_queue"])
+            assert isinstance(trigger.trigger, AzureServiceBusQueueTrigger)