AIP-82: Add RedisPubSubMessageQueueProvider

providers/redis/docs/index.rst

@@ -36,6 +36,7 @@
 
     Connection types <connections>
     Logging <logging/index>
+    Message Queues <message-queues>
     Triggers <triggers>
 
 .. toctree::

providers/redis/docs/message-queues.rst

@@ -0,0 +1,72 @@
+ .. 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.
+
+Redis Message Queue
+===================
+
+
+Redis Queue Provider
+--------------------
+
+Implemented by :class:`~airflow.providers.redis.queues.redis.RedisPubSubMessageQueueProvider`
+
+
+The Redis Queue Provider is a message queue provider that uses
+Redis as the underlying message queue system.
+It allows you to send and receive messages using Redis in your Airflow workflows.
+The provider supports Redis channels.
+
+The queue must be matching this regex:
+
+.. exampleinclude::/../src/airflow/providers/redis/queues/redis.py
+   :language: python
+   :dedent: 0
+   :start-after: [START queue_regexp]
+   :end-before: [END queue_regexp]
+
+Queue URI Format:
+
+.. code-block:: text
+
+   redis://<host>:<port>/<channel_list>
+
+Where:
+
+- ``host``: Redis server hostname
+- ``port``: Redis server port
+- ``channel_list``: Comma-separated list of Redis channels to subscribe to
+
+The ``queue`` parameter is used to configure the underlying
+:class:`~airflow.providers.redis.triggers.redis_await_message.AwaitMessageTrigger` class and
+passes all kwargs directly to the trigger constructor, if provided.
+
+Channels can also be specified via the Queue URI instead of the ``channels`` kwarg. The provider will extract channels from the URI as follows:
+
+.. exampleinclude:: /../src/airflow/providers/redis/queues/redis.py
+   :language: python
+   :dedent: 0
+   :start-after: [START extract_channels]
+   :end-before: [END extract_channels]
+
+
+Below is an example of how you can configure an Airflow DAG to be triggered by a message in Redis.
+
+.. exampleinclude:: /../tests/system/redis/example_dag_message_queue_trigger.py
+   :language: python
+   :dedent: 0
+   :start-after: [START howto_trigger_message_queue]
+   :end-before: [END howto_trigger_message_queue]

providers/redis/provider.yaml

@@ -68,6 +68,9 @@ operators:
     python-modules:
       - airflow.providers.redis.operators.redis_publish
 
+queues:
+  - airflow.providers.redis.queues.redis.RedisPubSubMessageQueueProvider
+
 sensors:
   - integration-name: Redis
     python-modules:

providers/redis/pyproject.toml

@@ -62,11 +62,19 @@ dependencies = [
     "redis>=4.5.2,!=4.5.5,!=5.0.2",
 ]
 
+# The optional dependencies should be modified in place in the generated file
+# Any change in the dependencies is preserved when the file is regenerated
+[project.optional-dependencies]
+"common.messaging" = [
+    "apache-airflow-providers-common-messaging>=1.0.3"
+]
+
 [dependency-groups]
 dev = [
     "apache-airflow",
     "apache-airflow-task-sdk",
     "apache-airflow-devel-common",
+    "apache-airflow-providers-common-messaging",
     # Additional devel dependencies (do not remove this line and add extra development dependencies)
 ]
 

providers/redis/src/airflow/providers/redis/get_provider_info.py

@@ -40,6 +40,7 @@ def get_provider_info():
                 "python-modules": ["airflow.providers.redis.operators.redis_publish"],
             }
         ],
+        "queues": ["airflow.providers.redis.queues.redis.RedisPubSubMessageQueueProvider"],
         "sensors": [
             {
                 "integration-name": "Redis",

providers/redis/src/airflow/providers/redis/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/redis/src/airflow/providers/redis/queues/redis.py

@@ -0,0 +1,91 @@
+# 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
+
+import re
+from typing import TYPE_CHECKING
+from urllib.parse import urlparse
+
+from airflow.providers.common.messaging.providers.base_provider import BaseMessageQueueProvider
+from airflow.providers.redis.triggers.redis_await_message import AwaitMessageTrigger
+
+if TYPE_CHECKING:
+    from airflow.triggers.base import BaseEventTrigger
+
+# [START queue_regexp]
+QUEUE_REGEXP = r"^redis\+pubsub://"
+# [END queue_regexp]
+
+
+class RedisPubSubMessageQueueProvider(BaseMessageQueueProvider):
+    """
+    Configuration for Redis integration with common-messaging.
+
+    It uses the ``redis+pubsub://`` URI scheme for identifying Redis queues.
+
+    **URI Format**:
+
+    .. code-block:: text
+
+        redis+pubsub://<host>:<port>/<channel_list>
+
+    Where:
+
+    * ``host``: Redis server hostname
+    * ``port``: Redis server port
+    * ``channel_list``: Comma-separated list of Redis channels to subscribe to
+
+    **Examples**:
+
+    .. code-block:: text
+
+        redis+pubsub://localhost:6379/my_channel
+
+    You can also provide ``channels`` directly in kwargs instead of in the URI.
+
+    .. code-block:: python
+
+        from airflow.providers.common.messaging.triggers.msg_queue import MessageQueueTrigger
+
+        trigger = MessageQueueTrigger(queue="redis+pubsub://localhost:6379/test")
+
+    For a complete example, see:
+    :mod:`tests.system.redis.example_dag_message_queue_trigger`
+    """
+
+    def queue_matches(self, queue: str) -> bool:
+        return bool(re.match(QUEUE_REGEXP, queue))
+
+    def trigger_class(self) -> type[BaseEventTrigger]:
+        return AwaitMessageTrigger  # type: ignore[return-value]
+
+    def trigger_kwargs(self, queue: str, **kwargs) -> dict:
+        # [START extract_channels]
+        # Parse the queue URI
+        parsed = urlparse(queue)
+        # Extract channels (after host and port)
+        # parsed.path starts with a '/', so strip it
+        raw_channels = parsed.path.lstrip("/")
+        channels = raw_channels.split(",") if raw_channels else []
+        # [END extract_channels]
+
+        if not channels and "channels" not in kwargs:
+            raise ValueError(
+                "channels is required in RedisPubSubMessageQueueProvider kwargs or provide them in the queue URI"
+            )
+
+        return {} if "channels" in kwargs else {"channels": channels}

providers/redis/tests/integration/redis/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/redis/tests/integration/redis/queues/test_redis_pubsub_message_queue.py

@@ -0,0 +1,60 @@
+# 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
+
+import time
+
+import pytest
+
+from airflow.providers.redis.hooks.redis import RedisHook
+from airflow.providers.redis.queues.redis import RedisPubSubMessageQueueProvider
+
+
+@pytest.mark.integration("redis")
+class TestRedisPubSubMessageQueueProviderIntegration:
+    def setup_method(self):
+        self.redis_hook = RedisHook(redis_conn_id="redis_default")
+        self.redis = self.redis_hook.get_conn()
+        self.provider = RedisPubSubMessageQueueProvider()
+        self.channel = "test_pubsub_channel"
+
+    def test_pubsub_send_and_receive(self):
+        pubsub = self.redis.pubsub()
+        pubsub.subscribe(self.channel)
+
+        test_message = "airflow-pubsub-integration-message"
+        self.redis.publish(self.channel, test_message)
+
+        received = None
+        for _ in range(10):
+            message = pubsub.get_message()
+            if message and message["type"] == "message":
+                received = message["data"]
+                break
+            time.sleep(0.1)
+
+        assert received == test_message.encode(), f"Expected {test_message!r}, got {received!r}"
+
+        pubsub.unsubscribe(self.channel)
+
+    def test_queue_matches(self):
+        assert self.provider.queue_matches(f"redis+pubsub://localhost:6379/{self.channel}")
+
+    def test_trigger_kwargs(self):
+        uri = f"redis+pubsub://localhost:6379/{self.channel}"
+        kwargs = self.provider.trigger_kwargs(uri)
+        assert kwargs == {"channels": [self.channel]}

providers/redis/tests/system/redis/example_dag_message_queue_trigger.py

@@ -0,0 +1,38 @@
+# 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
+
+# [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 import DAG, Asset, AssetWatcher
+
+# Define a trigger that listens to an external message queue (Redis in this case)
+trigger = MessageQueueTrigger(queue="redis+pubsub://localhost:6379/test")
+
+# Define an asset that watches for messages on the queue
+asset = Asset("redis_queue_asset_1", watchers=[AssetWatcher(name="redis_watcher_1", trigger=trigger)])
+
+with DAG(dag_id="example_redis_watcher_1", schedule=[asset]) as dag:
+    EmptyOperator(task_id="task_1")
+# [END howto_trigger_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/redis/tests/unit/redis/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.