diff --git a/docs/spelling_wordlist.txt b/docs/spelling_wordlist.txt index cc9767243a79a..50265b5d6a545 100644 --- a/docs/spelling_wordlist.txt +++ b/docs/spelling_wordlist.txt @@ -94,6 +94,7 @@ arraysize artifactId Asana asana +asb asc ascii asciiart diff --git a/providers/microsoft/azure/docs/index.rst b/providers/microsoft/azure/docs/index.rst index c1c3168cc26fb..6aa0bfe4483fe 100644 --- a/providers/microsoft/azure/docs/index.rst +++ b/providers/microsoft/azure/docs/index.rst @@ -35,6 +35,7 @@ :caption: Guides Connection types + Message queues Operators Transfers Filesystems @@ -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 `_ ``amazon`` -`apache-airflow-providers-common-compat `_ ``common.compat`` -`apache-airflow-providers-oracle `_ ``oracle`` -`apache-airflow-providers-sftp `_ ``sftp`` -================================================================================================================== ================= +======================================================================================================================== ==================== +Dependent package Extra +======================================================================================================================== ==================== +`apache-airflow-providers-amazon `_ ``amazon`` +`apache-airflow-providers-common-compat `_ ``common.compat`` +`apache-airflow-providers-common-messaging `_ ``common.messaging`` +`apache-airflow-providers-oracle `_ ``oracle`` +`apache-airflow-providers-sftp `_ ``sftp`` +======================================================================================================================== ==================== Downloading official packages ----------------------------- diff --git a/providers/microsoft/azure/docs/message-queues/index.rst b/providers/microsoft/azure/docs/message-queues/index.rst new file mode 100644 index 0000000000000..a0a9fba9ac2b5 --- /dev/null +++ b/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 ` diff --git a/providers/microsoft/azure/provider.yaml b/providers/microsoft/azure/provider.yaml index 2b85758f432bf..6cb3921ebcdab 100644 --- a/providers/microsoft/azure/provider.yaml +++ b/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 diff --git a/providers/microsoft/azure/pyproject.toml b/providers/microsoft/azure/pyproject.toml index 50eab1ad83153..00ca93d806339 100644 --- a/providers/microsoft/azure/pyproject.toml +++ b/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) diff --git a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/get_provider_info.py b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/get_provider_info.py index 1966efa4fcb59..356ea5d13ef5c 100644 --- a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/get_provider_info.py +++ b/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", diff --git a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/queues/__init__.py b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/queues/__init__.py new file mode 100644 index 0000000000000..13a83393a9124 --- /dev/null +++ b/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. diff --git a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/queues/asb.py b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/queues/asb.py new file mode 100644 index 0000000000000..867be6693f7d1 --- /dev/null +++ b/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 diff --git a/providers/microsoft/azure/tests/system/microsoft/azure/example_event_schedule_asb.py b/providers/microsoft/azure/tests/system/microsoft/azure/example_event_schedule_asb.py new file mode 100644 index 0000000000000..9d95040ab33b1 --- /dev/null +++ b/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) diff --git a/providers/microsoft/azure/tests/unit/microsoft/azure/queues/__init__.py b/providers/microsoft/azure/tests/unit/microsoft/azure/queues/__init__.py new file mode 100644 index 0000000000000..13a83393a9124 --- /dev/null +++ b/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. diff --git a/providers/microsoft/azure/tests/unit/microsoft/azure/queues/test_asb.py b/providers/microsoft/azure/tests/unit/microsoft/azure/queues/test_asb.py new file mode 100644 index 0000000000000..01e246bb23411 --- /dev/null +++ b/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)