Skip to content

Move MESSAGE_QUEUE_PROVIDERS array to where it belongs - to msq_queue #51774

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
potiuk merged 1 commit into from
Jun 16, 2025
Merged

Move MESSAGE_QUEUE_PROVIDERS array to where it belongs - to msq_queue #51774

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 0 additions & 15 deletions providers/common/messaging/src/airflow/providers/common/messaging/providers/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,23 +16,8 @@
# under the License.
from __future__ import annotations

import importlib

from airflow.providers_manager import ProvidersManager
from airflow.utils.deprecation_tools import add_deprecated_classes

providers_manager = ProvidersManager()
providers_manager.initialize_providers_queues()


def create_class_by_name(name: str):
module_name, class_name = name.rsplit(".", 1)
module = importlib.import_module(module_name)
return getattr(module, class_name)


MESSAGE_QUEUE_PROVIDERS = [create_class_by_name(name)() for name in providers_manager.queue_class_names]

__deprecated_classes = {
"sqs": {
"SqsMessageQueueProvider": "airflow.providers.amazon.aws.queues.sqs.SqsMessageQueueProvider",
Expand Down
15 changes: 14 additions & 1 deletion providers/common/messaging/src/airflow/providers/common/messaging/triggers/msg_queue.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,13 +17,26 @@
# under the License.
from __future__ import annotations

import importlib
from collections.abc import AsyncIterator
from functools import cached_property
from typing import Any

from airflow.providers.common.messaging.providers import MESSAGE_QUEUE_PROVIDERS
from airflow.providers_manager import ProvidersManager
from airflow.triggers.base import BaseEventTrigger, TriggerEvent

providers_manager = ProvidersManager()
providers_manager.initialize_providers_queues()


def create_class_by_name(name: str):
module_name, class_name = name.rsplit(".", 1)
module = importlib.import_module(module_name)
return getattr(module, class_name)


MESSAGE_QUEUE_PROVIDERS = [create_class_by_name(name)() for name in providers_manager.queue_class_names]


class MessageQueueTrigger(BaseEventTrigger):
"""
Expand Down
3 changes: 2 additions & 1 deletion providers/common/messaging/tests/unit/common/messaging/triggers/test_msg_queue.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,8 @@


@mock.patch(
"airflow.providers.common.messaging.providers.MESSAGE_QUEUE_PROVIDERS", new_callable=mock.PropertyMock
"airflow.providers.common.messaging.triggers.msg_queue.MESSAGE_QUEUE_PROVIDERS",
new_callable=mock.PropertyMock,
)
def test_provider_integrations(_):
trigger = MessageQueueTrigger(queue="any queue")
Expand Down