You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by po...@apache.org on 2025/04/14 10:47:03 UTC
(airflow) branch main updated: Move SQS message queue code example from core to provider docs (#49208)
This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new bc45acf4ccb Move SQS message queue code example from core to provider docs (#49208)
bc45acf4ccb is described below
commit bc45acf4ccb7db87b410f3d03a7f8be3364ec5f4
Author: Tamara Janina Fingerlin <90...@users.noreply.github.com>
AuthorDate: Mon Apr 14 12:46:57 2025 +0200
Move SQS message queue code example from core to provider docs (#49208)
* Move SQS message queue code example from core to provider docs
* fix code snippet
* Fix cross link
* Update providers/common/messaging/docs/triggers.rst
---
.../authoring-and-scheduling/event-scheduling.rst | 35 +---------------------
providers/common/messaging/docs/triggers.rst | 18 +++++++++--
.../messaging/example_message_queue_trigger.py | 16 +++++-----
3 files changed, 24 insertions(+), 45 deletions(-)
diff --git a/airflow-core/docs/authoring-and-scheduling/event-scheduling.rst b/airflow-core/docs/authoring-and-scheduling/event-scheduling.rst
index be6070c53e5..e6bcbe4ae86 100644
--- a/airflow-core/docs/authoring-and-scheduling/event-scheduling.rst
+++ b/airflow-core/docs/authoring-and-scheduling/event-scheduling.rst
@@ -34,40 +34,7 @@ message queue, and triggers an asset update when a relevant event occurs.
The ``watchers`` parameter in the ``Asset`` definition allows you to associate multiple ``AssetWatcher`` instances with an
asset, enabling it to respond to various event sources.
-Example: Triggering a DAG from an external message queue
---------------------------------------------------------
-
-Below is an example of how you can configure an Airflow DAG to be triggered by an external message queue, such as AWS
-SQS:
-
-.. code-block:: python
-
- from airflow.sdk import Asset, AssetWatcher
- from airflow.providers.common.msgq.triggers.msg_queue import MessageQueueTrigger
- from airflow.sdk import DAG
- from datetime import datetime
-
- # Define a trigger that listens to an external message queue (AWS SQS in this case)
- trigger = MessageQueueTrigger(queue="https://sqs.us-east-1.amazonaws.com/0123456789/my-queue")
-
- # Define an asset that watches for messages on the queue
- asset = Asset("sqs_queue_asset", watchers=[AssetWatcher(name="sqs_watcher", trigger=trigger)])
-
- # Define the DAG that will be triggered when the asset is updated
- with DAG(dag_id="event_driven_dag", schedule=[asset], catchup=False) as dag:
- ...
-
-How it works
-------------
-1. **Message Queue Trigger**: The ``MessageQueueTrigger`` listens for messages from an external queue
-(e.g., AWS SQS, Kafka, or another messaging system).
-
-2. **Asset and Watcher**: The ``Asset`` abstracts the external entity, the SQS 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).
+See the :doc:`common.messaging provider docs <apache-airflow-providers-common-messaging:triggers>` for more information and examples.
Supported triggers for event-driven scheduling
----------------------------------------------
diff --git a/providers/common/messaging/docs/triggers.rst b/providers/common/messaging/docs/triggers.rst
index 8e249fd20f7..8d02836f4d9 100644
--- a/providers/common/messaging/docs/triggers.rst
+++ b/providers/common/messaging/docs/triggers.rst
@@ -29,11 +29,25 @@ queue. Parameters of the trigger are:
- ``queue`` - the queue identifier
-Additional parameters can be provided depending on the queue provider.
+Additional parameters can be provided depending on the queue provider. Connections needs to be provided with the relevant
+default connection ID, for example, when connecting to a queue in AWS SQS, the connection ID should be
+``aws_default``.
-The example below shows how to schedule a DAG using MessageQueueTrigger.
+Below is an example of how you can configure an Airflow DAG to be triggered by a message in Amazon SQS.
.. exampleinclude:: /../tests/system/common/messaging/example_message_queue_trigger.py
:language: python
:start-after: [START howto_trigger_message_queue]
:end-before: [END howto_trigger_message_queue]
+
+How it works
+------------
+1. **Message Queue Trigger**: The ``MessageQueueTrigger`` listens for messages from an external queue
+(e.g., AWS SQS, Kafka, or another messaging system).
+
+2. **Asset and Watcher**: The ``Asset`` abstracts the external entity, the SQS 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).
diff --git a/providers/common/messaging/tests/system/common/messaging/example_message_queue_trigger.py b/providers/common/messaging/tests/system/common/messaging/example_message_queue_trigger.py
index 2f7b9128e82..7eba4dc35aa 100644
--- a/providers/common/messaging/tests/system/common/messaging/example_message_queue_trigger.py
+++ b/providers/common/messaging/tests/system/common/messaging/example_message_queue_trigger.py
@@ -16,20 +16,18 @@
# under the License.
from __future__ import annotations
-from airflow.models.dag import DAG
+# [START howto_trigger_message_queue]
from airflow.providers.common.messaging.triggers.msg_queue import MessageQueueTrigger
from airflow.providers.standard.operators.empty import EmptyOperator
-from airflow.sdk.definitions.asset import Asset, AssetWatcher
+from airflow.sdk import DAG, Asset, AssetWatcher
-# [START howto_trigger_message_queue]
+# Define a trigger that listens to an external message queue (AWS SQS in this case)
trigger = MessageQueueTrigger(queue="https://sqs.us-east-1.amazonaws.com/0123456789/my-queue")
-asset = Asset("sqs_asset", watchers=[AssetWatcher(name="sqs_asset_watcher", trigger=trigger)])
-with DAG(
- dag_id="example_msgq_watcher",
- schedule=[asset],
- catchup=False,
-) as dag:
+# Define an asset that watches for messages on the queue
+asset = Asset("sqs_queue_asset", watchers=[AssetWatcher(name="sqs_watcher", trigger=trigger)])
+
+with DAG(dag_id="example_msgq_watcher", schedule=[asset]) as dag:
EmptyOperator(task_id="task")
# [END howto_trigger_message_queue]