Copilot commented on code in PR #62790:
URL: https://github.com/apache/airflow/pull/62790#discussion_r2984869099
##########
airflow-core/docs/extra-packages-ref.rst:
##########
@@ -339,6 +339,8 @@ Some of those enable Airflow to use executors to run tasks
with them - other tha
+---------------------+-----------------------------------------------------+-----------------------------------------------------------------+----------------------------------------------+
| influxdb | ``pip install 'apache-airflow[influxdb]'`` |
Influxdb operators and hook |
|
+---------------------+-----------------------------------------------------+-----------------------------------------------------------------+----------------------------------------------+
+| ibm-mq | ``pip install 'apache-airflow[ibm-mq]'`` |
IBM MQ hooks and operators |
|
++---------------------+-----------------------------------------------------+-----------------------------------------------------------------+----------------------------------------------+
Review Comment:
This extra-packages table entry says “hooks and operators”, but the provider
currently defines hooks/triggers/queue provider and no operators. Please update
the description to match what the package actually ships (or add the missing
operators if intended).
##########
providers/ibm/mq/tests/system/ibm/mq/example_dag_message_queue_trigger.py:
##########
@@ -0,0 +1,48 @@
+# 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.sdk import DAG, Asset, AssetWatcher, task
+
+# Define a trigger that listens to an external message queue (IBM MQ in this
case)
+trigger = MessageQueueTrigger(
+ queue="mq://mq_default/MY.QUEUE.NAME",
+)
+
+mq_topic_asset = Asset(
+ "mq_topic_asset",
+ watchers=[AssetWatcher(name="mq_watcher", trigger=trigger)],
+)
+
+with DAG(dag_id="example_ibm_mq_watcher", schedule=[mq_topic_asset]) as dag:
+
+ @task
+ def process_message(**context):
+ for event in context["triggering_asset_events"][mq_topic_asset]:
+ # Get the message from the TriggerEvent payload
+ print("Processing event: ", event)
+ payload = event["payload"]
+ print("Actual payload: ", payload)
Review Comment:
The example treats each triggering event as a dict with a `"payload"` key,
but `MessageQueueTrigger` forwards the provider trigger’s `TriggerEvent`
payload directly (Kafka and this IBM MQ trigger both yield a string payload).
As written, `event["payload"]` will fail; update the example to handle the
actual payload type (or change the trigger to yield a dict if that’s the
intended contract).
##########
providers/ibm/mq/src/airflow/providers/ibm/mq/hooks/mq.py:
##########
@@ -0,0 +1,404 @@
+#
+# 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 asyncio
+import json
+import threading
+from contextlib import contextmanager, suppress
+from typing import Any
+
+from asgiref.sync import sync_to_async
+
+from airflow.providers.common.compat.sdk import BaseHook
+
+# Backoff parameters for transient consume failures
+_BACKOFF_BASE: float = 1.0
+_BACKOFF_MAX: float = 60.0
+_BACKOFF_FACTOR: float = 2.0
+
+
+class IBMMQHook(BaseHook):
+ """
+ Interact with IBM MQ queue managers to consume and produce messages.
+
+ This hook wraps the ``ibmmq`` C client and manages connection
+ lifecycle, queue open/close, and message serialization. Both synchronous
+ (context-manager) and asynchronous (``consume`` / ``produce``) interfaces
+ are provided.
+
+ Connection parameters (host, port, login, password) are read from the
+ Airflow connection identified by *conn_id*. ``queue_manager``,
``channel``,
+ and ``open_options`` can be supplied either as constructor arguments or via
+ the connection's *extra* JSON — constructor arguments take precedence.
+
+ :param conn_id: Airflow connection ID for the IBM MQ instance.
+ Defaults to ``"mq_default"``.
+ :param queue_manager: Name of the IBM MQ queue manager to connect to.
+ If not provided, the value is read from the ``queue_manager`` key in
+ the connection's *extra* JSON.
+ :param channel: MQ channel name used for the connection.
+ If not provided, the value is read from the ``channel`` key in the
+ connection's *extra* JSON.
+ :param open_options: Integer bitmask of ``MQOO_*`` open options passed
+ when opening a queue (e.g.,
+ ``ibmmq.CMQC.MQOO_INPUT_SHARED | ibmmq.CMQC.MQOO_FAIL_IF_QUIESCING``).
+ If not provided, the value is resolved from the ``open_options`` key
+ in the connection's *extra* JSON, falling back to
+ ``MQOO_INPUT_SHARED``.
+ """
+
+ conn_name_attr = "conn_id"
+ default_conn_name = "mq_default"
+ conn_type = "mq"
+ hook_name = "IBM MQ"
+ default_open_options = "MQOO_INPUT_SHARED"
+
+ def __init__(
+ self,
+ conn_id: str = default_conn_name,
+ queue_manager: str | None = None,
+ channel: str | None = None,
+ open_options: int | None = None,
+ ):
+ super().__init__()
+ self.conn_id = conn_id
+ self.queue_manager = queue_manager
+ self.channel = channel
+ self.open_options = open_options
+ self._conn = None
+
+ @classmethod
+ def get_ui_field_behaviour(cls) -> dict[str, Any]:
+ """Return custom UI field behaviour for IBM MQ Connection."""
+ return {
+ "hidden_fields": ["schema"],
+ "placeholders": {
+ "host": "mq.example.com",
+ "port": "1414",
+ "login": "app_user",
+ "extra": json.dumps(
+ {
+ "queue_manager": "QM1",
+ "channel": "DEV.APP.SVRCONN",
+ "open_options": cls.default_open_options,
+ },
+ indent=2,
+ ),
+ },
+ }
+
+ @classmethod
+ def get_open_options_flags(cls, open_options: int) -> list[str]:
+ """
+ Return the symbolic MQ open option flags set in a given bitmask.
+
+ Each flag corresponds to a constant in ``ibmmq.CMQC`` that starts with
``MQOO_``.
+
+ :param open_options: The integer bitmask used when opening an MQ queue
+ (e.g., ``MQOO_INPUT_EXCLUSIVE |
MQOO_FAIL_IF_QUIESCING``).
+
+ :return: A list of the names of the MQ open flags that are set in the
bitmask.
+ For example, ``['MQOO_INPUT_EXCLUSIVE',
'MQOO_FAIL_IF_QUIESCING']``.
+
+ Example:
+ >>> open_options = ibmmq.CMQC.MQOO_INPUT_SHARED |
ibmmq.CMQC.MQOO_FAIL_IF_QUIESCING
+ >>> cls.get_open_options_flags(open_options)
+ ['MQOO_INPUT_SHARED', 'MQOO_FAIL_IF_QUIESCING']
+ """
+ import ibmmq
+
+ return [
+ name
+ for name, value in vars(ibmmq.CMQC).items()
+ if name.startswith("MQOO_") and (open_options & value)
+ ]
+
+ @staticmethod
+ def _connect(queue_manager: str, channel: str, conn_info: str, csp):
+ """
+ Connect to the IBM MQ queue manager.
+
+ Connection errors from the C client are caught and re-raised as a
+ :class:`ConnectionError` with a human-readable message.
+
+ :return: IBM MQ connection object
+ """
+ import ibmmq
+
+ try:
+ return ibmmq.connect(queue_manager, channel, conn_info, csp=csp)
+ except (ibmmq.MQMIError, ibmmq.PYIFError) as e:
+ raise ConnectionError(
+ f"Failed to connect to IBM MQ queue manager '{queue_manager}' "
+ f"at {conn_info} on channel '{channel}': {e}"
+ ) from e
+
+ @contextmanager
+ def get_conn(self):
+ """
+ Sync context manager for IBM MQ connection lifecycle.
+
+ Must be called from the executor thread (not the event loop thread).
+ Retrieves the Airflow connection, extracts MQ parameters, and manages
+ the IBM MQ connection lifecycle.
+
+ :yield: IBM MQ connection object
+ """
+ import ibmmq
+
+ connection = BaseHook.get_connection(self.conn_id)
+ config = connection.extra_dejson
+ queue_manager = self.queue_manager or config.get("queue_manager")
+ channel = self.channel or config.get("channel")
+
+ if not queue_manager:
+ raise ValueError("queue_manager must be set in Connection extra
config or hook init")
+ if not channel:
+ raise ValueError("channel must be set in Connection extra config
or hook init")
+
+ if not self.open_options:
+ self.open_options = getattr(
+ ibmmq.CMQC,
+ config.get("open_options", self.default_open_options),
+ ibmmq.CMQC.MQOO_INPUT_EXCLUSIVE,
+ )
Review Comment:
`get_conn()` resolves `open_options` from connection extras, but the
fallback passed to `getattr()` uses `MQOO_INPUT_EXCLUSIVE` even though the
docstring/defaults indicate `MQOO_INPUT_SHARED`. This changes default
queue-open behavior (exclusive vs shared) and can block other consumers; please
align the fallback constant with the documented default.
##########
providers/ibm/mq/tests/unit/ibm/mq/hooks/test_mq.py:
##########
@@ -0,0 +1,248 @@
+# 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 asyncio
+from unittest import mock
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from airflow.models import Connection
+from airflow.providers.ibm.mq.hooks.mq import (
+ _BACKOFF_BASE,
+ _BACKOFF_FACTOR,
+ _BACKOFF_MAX,
+ IBMMQHook,
+)
+
+MQ_PAYLOAD = """RFH x"MQSTR <mcd><Msd>jms_map</Msd></mcd>
<jms><Dst>topic://localhost/topic</Dst><Tms>1772121947476</Tms><Dlv>2</Dlv><Uci
dt='bin.hex'>414D5143514D4941303054202020202069774D7092F81057</Uci></jms>L<usr><XMSC_CLIENT_ID>local</XMSC_CLIENT_ID><release>26.01.00</release></usr>
4<mqps><Top>topic</Top></mqps> {}"""
+
+
+def mq_connection():
+ """Create a test MQ connection object."""
+ return Connection(
+ conn_id="mq_conn",
+ conn_type="mq",
+ host="mq.example.com",
+ login="user",
+ password="pass",
+ port=1414,
+ extra='{"queue_manager": "QM1", "channel": "DEV.APP.SVRCONN"}',
+ )
+
+
[email protected]
+def mock_get_connection():
+ """Fixture that mocks BaseHook.get_connection to return a test
connection."""
+ with patch("airflow.providers.ibm.mq.hooks.mq.BaseHook.get_connection") as
mock_conn:
+ mock_conn.return_value = mq_connection()
+ yield mock_conn
+
+
[email protected]
+def patch_sync_to_async():
+ """Patch sync_to_async to call the wrapped function directly for
testing."""
+
+ def sync_to_async(func):
+ """Wrap a sync function so it can be awaited directly."""
+
+ async def wrapper(*args, **kwargs):
+ return func(*args, **kwargs)
+
+ return wrapper
+
+ with patch("airflow.providers.ibm.mq.hooks.mq.sync_to_async",
side_effect=sync_to_async):
+ yield
+
+
+def fake_get(*args, **kwargs):
+ import ibmmq
+
+ raise ibmmq.MQMIError("connection broken",
reason=ibmmq.CMQC.MQRC_CONNECTION_BROKEN)
+
+
[email protected]
+class TestIBMMQHook:
+ """Tests for the IBM MQ hook."""
+
+ @pytest.fixture(autouse=True)
+ def setup_connections(self, create_connection_without_db):
+ create_connection_without_db(
+ Connection(
+ conn_id="mq_conn",
+ conn_type="mq",
+ host="mq.example.com",
+ login="user",
+ password="pass",
+ port=1414,
+ extra='{"queue_manager": "QM1", "channel": "DEV.APP.SVRCONN"}',
+ )
+ )
+ self.hook = IBMMQHook("mq_conn")
+
+ @patch("ibmmq.connect")
+ @patch("ibmmq.Queue")
+ async def test_aconsume_message(
+ self, mock_queue_class, mock_connect, mock_get_connection,
patch_sync_to_async
+ ):
+ """Test consuming a single message."""
+
+ mock_qmgr = MagicMock()
+ mock_connect.return_value = mock_qmgr
+
+ mock_queue = MagicMock()
+ mock_queue_class.return_value = mock_queue
+ mock_queue.get.return_value = MQ_PAYLOAD.format("test
message").encode()
Review Comment:
These tests use `MagicMock()` for MQ connection/queue objects without
`spec`/`autospec`, which can hide attribute/usage bugs (e.g. typos on MQ API
calls) because mocks will accept anything. Consider using `autospec=True` in
the patches or `MagicMock(spec=...)` for the returned objects.
##########
providers/ibm/mq/README.rst:
##########
@@ -0,0 +1,92 @@
+
+.. 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.
+
+.. NOTE! THIS FILE IS AUTOMATICALLY GENERATED AND WILL BE OVERWRITTEN!
+
+.. IF YOU WANT TO MODIFY TEMPLATE FOR THIS FILE, YOU SHOULD MODIFY THE TEMPLATE
+ ``PROVIDER_README_TEMPLATE.rst.jinja2`` IN the
``dev/breeze/src/airflow_breeze/templates`` DIRECTORY
+
+Package ``apache-airflow-providers-ibm-mq``
+
+Release: ``0.1.0``
+
+
+`IBM MQ <https://www.ibm.com/products/mq/>`__
+
+
+Provider package
+----------------
+
+This is a provider package for ``ibm.mq`` provider. All classes for this
provider package
+are in ``airflow.providers.ibm.mq`` python package.
+
+You can find package information and changelog for the provider
+in the `documentation
<https://airflow.apache.org/docs/apache-airflow-providers-ibm-mq/0.1.0/>`_.
+
+Installation
+------------
+
+You can install this package on top of an existing Airflow 2 installation
+(see ``Requirements`` below for the minimum Airflow version supported) via
+
+``pip install apache-airflow-providers-ibm-mq``
+
+This installs only the provider package. To use the IBM MQ operators at
+runtime you also need the ``ibmmq`` dependency, which can be installed via
+the provider extra:
+
+``pip install apache-airflow-providers-ibm-mq[ibmmq]``
+
+The ``ibmmq`` extra installs the Python wrapper for the IBM MQ client that is
required by the provider hooks and operators.
+
+Note that the `ibmmq <https://github.com/ibm-messaging/mq-mqi-python/>`_
Python package requires the native `IBM MQ Redistributable Client
<https://www.ibm.com/docs/en/ibm-mq/9.4.x?topic=overview-redistributable-mq-clients>`_
libraries to be installed on the system.
+
+Refer to the IBM MQ documentation for installation instructions for your
platform.
+
+The package supports the following python versions: 3.9,3.10,3.11,3.12
+
+Requirements
+------------
+
+=============================================
=====================================
+PIP package Version required
+=============================================
=====================================
+``apache-airflow`` ``>=2.11.0``
+``apache-airflow-providers-common-messaging`` ``>=2.0.0``
+``importlib-resources`` ``>=1.3``
+=============================================
=====================================
Review Comment:
The README claims support for Python 3.9 and lists runtime requirements
(`apache-airflow-providers-common-messaging`, `importlib-resources`) that don’t
match this provider’s `pyproject.toml` (Python >=3.10; no required
common-messaging/importlib-resources). Please regenerate/fix the README so the
supported versions and requirements reflect the actual package metadata.
##########
providers/ibm/mq/src/airflow/providers/ibm/mq/queues/mq.py:
##########
@@ -0,0 +1,86 @@
+#
+# 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, Any
+from urllib.parse import urlparse
+
+from airflow.providers.common.messaging.providers.base_provider import
BaseMessageQueueProvider
+from airflow.providers.ibm.mq.triggers.mq import AwaitMessageTrigger
Review Comment:
`BaseMessageQueueProvider` is imported unconditionally. Other message-queue
providers guard this import and raise `AirflowOptionalProviderFeatureException`
when `common.messaging` isn’t installed, so importing this module doesn’t crash
Airflow in environments without that optional provider.
##########
providers/ibm/mq/tests/unit/ibm/mq/__init__.py:
##########
@@ -0,0 +1,51 @@
+# 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 importlib.util
+
+if not importlib.util.find_spec("ibmmq"):
+ import sys
+ from unittest.mock import MagicMock
+
+ class MQMIError(Exception):
+ def __init__(self, msg="", reason=None):
+ super().__init__(msg)
+ self.reason = reason
+
+ class PYIFError(Exception):
+ def __init__(self, msg="", reason=None):
+ super().__init__(msg)
+ self.reason = reason
+
+ fake_ibmmq = MagicMock()
+ fake_ibmmq.CMQC.MQRC_NO_MSG_AVAILABLE = 2033
+ fake_ibmmq.CMQC.MQRC_CONNECTION_BROKEN = 2009
+ fake_ibmmq.CMQC.MQGMO_WAIT = 1
+ fake_ibmmq.CMQC.MQGMO_NO_SYNCPOINT = 2
+ fake_ibmmq.CMQC.MQGMO_CONVERT = 4
+ fake_ibmmq.MQMIError = MQMIError
+ fake_ibmmq.PYIFError = PYIFError
+ fake_ibmmq.OD = MagicMock()
+ fake_ibmmq.MD = MagicMock()
+ fake_ibmmq.GMO = MagicMock()
+ fake_ibmmq.CSP = MagicMock()
+ fake_ibmmq.Queue = MagicMock()
+ fake_ibmmq.connect = MagicMock()
+ fake_ibmmq.RFH2 = MagicMock()
+
+ sys.modules["ibmmq"] = fake_ibmmq
Review Comment:
This package `__init__.py` mutates `sys.modules["ibmmq"]` at import time,
which is a global side effect that can leak into unrelated tests and mask
missing-dependency behavior. Prefer using a pytest fixture (e.g.,
`monkeypatch.setitem(sys.modules, ...)`) scoped to this provider’s tests, or
patch at the test-module level instead of during package import.
##########
providers/ibm/mq/docs/changelog.rst:
##########
@@ -0,0 +1,49 @@
+ .. 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.
+
+
+.. NOTE TO CONTRIBUTORS:
+ Please, only add notes to the Changelog just below the "Changelog" header
when there are some breaking changes
+ and you want to add an explanation to the users on how they are supposed to
deal with them.
+ The changelog is updated and maintained semi-automatically by release
manager.
+
+``apache-airflow-providers-ibm-mq``
+
+Changelog
+---------
+
+0.1.0
+.....
+
+.. note::
+ This release of provider is only available for Airflow 2.10+ as explained
in the
Review Comment:
The changelog note says this provider is only available for Airflow 2.10+,
but the provider package runtime guard requires Airflow 2.11.0+ (see
`airflow.providers.ibm.mq.__init__`). Please make these consistent to avoid
misleading users.
```suggestion
This release of provider is only available for Airflow 2.11.0+ as
explained in the
```
##########
providers/ibm/mq/docs/changelog.rst:
##########
@@ -0,0 +1,49 @@
+ .. 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.
+
+
+.. NOTE TO CONTRIBUTORS:
+ Please, only add notes to the Changelog just below the "Changelog" header
when there are some breaking changes
+ and you want to add an explanation to the users on how they are supposed to
deal with them.
+ The changelog is updated and maintained semi-automatically by release
manager.
+
+``apache-airflow-providers-ibm-mq``
+
+Changelog
+---------
+
+0.1.0
+.....
+
+.. note::
+ This release of provider is only available for Airflow 2.10+ as explained
in the
+ Apache Airflow providers support policy
<https://github.com/apache/airflow/blob/main/PROVIDERS.rst#minimum-supported-version-of-airflow-for-community-managed-providers>_.
+
+Misc
+~~~~
+
+* ``Bump min Airflow version in providers to 2.10 (#49843)``
+
+.. Below changes are excluded from the changelog. Move them to
+ appropriate section above if needed. Do not delete the lines(!):
+ * ``Update description of provider.yaml dependencies (#50231)``
+ * ``Avoid committing history for providers (#49907)``
+
+0.1.0
+.....
+
+Initial version of the provider.
Review Comment:
`0.1.0` appears twice as a top-level release heading, which will render
confusing/duplicated sections in the published changelog. Please remove the
duplicate heading/content or merge into a single `0.1.0` section.
##########
providers/ibm/mq/src/airflow/providers/ibm/mq/hooks/mq.py:
##########
@@ -0,0 +1,404 @@
+#
+# 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 asyncio
+import json
+import threading
+from contextlib import contextmanager, suppress
+from typing import Any
+
+from asgiref.sync import sync_to_async
+
+from airflow.providers.common.compat.sdk import BaseHook
+
+# Backoff parameters for transient consume failures
+_BACKOFF_BASE: float = 1.0
+_BACKOFF_MAX: float = 60.0
+_BACKOFF_FACTOR: float = 2.0
+
+
+class IBMMQHook(BaseHook):
+ """
+ Interact with IBM MQ queue managers to consume and produce messages.
+
+ This hook wraps the ``ibmmq`` C client and manages connection
+ lifecycle, queue open/close, and message serialization. Both synchronous
+ (context-manager) and asynchronous (``consume`` / ``produce``) interfaces
+ are provided.
+
+ Connection parameters (host, port, login, password) are read from the
+ Airflow connection identified by *conn_id*. ``queue_manager``,
``channel``,
+ and ``open_options`` can be supplied either as constructor arguments or via
+ the connection's *extra* JSON — constructor arguments take precedence.
+
+ :param conn_id: Airflow connection ID for the IBM MQ instance.
+ Defaults to ``"mq_default"``.
+ :param queue_manager: Name of the IBM MQ queue manager to connect to.
+ If not provided, the value is read from the ``queue_manager`` key in
+ the connection's *extra* JSON.
+ :param channel: MQ channel name used for the connection.
+ If not provided, the value is read from the ``channel`` key in the
+ connection's *extra* JSON.
+ :param open_options: Integer bitmask of ``MQOO_*`` open options passed
+ when opening a queue (e.g.,
+ ``ibmmq.CMQC.MQOO_INPUT_SHARED | ibmmq.CMQC.MQOO_FAIL_IF_QUIESCING``).
+ If not provided, the value is resolved from the ``open_options`` key
+ in the connection's *extra* JSON, falling back to
+ ``MQOO_INPUT_SHARED``.
+ """
+
+ conn_name_attr = "conn_id"
+ default_conn_name = "mq_default"
+ conn_type = "mq"
+ hook_name = "IBM MQ"
+ default_open_options = "MQOO_INPUT_SHARED"
+
+ def __init__(
+ self,
+ conn_id: str = default_conn_name,
+ queue_manager: str | None = None,
+ channel: str | None = None,
+ open_options: int | None = None,
+ ):
+ super().__init__()
+ self.conn_id = conn_id
+ self.queue_manager = queue_manager
+ self.channel = channel
+ self.open_options = open_options
+ self._conn = None
+
+ @classmethod
+ def get_ui_field_behaviour(cls) -> dict[str, Any]:
+ """Return custom UI field behaviour for IBM MQ Connection."""
+ return {
+ "hidden_fields": ["schema"],
+ "placeholders": {
+ "host": "mq.example.com",
+ "port": "1414",
+ "login": "app_user",
+ "extra": json.dumps(
+ {
+ "queue_manager": "QM1",
+ "channel": "DEV.APP.SVRCONN",
+ "open_options": cls.default_open_options,
+ },
+ indent=2,
+ ),
+ },
+ }
+
+ @classmethod
+ def get_open_options_flags(cls, open_options: int) -> list[str]:
+ """
+ Return the symbolic MQ open option flags set in a given bitmask.
+
+ Each flag corresponds to a constant in ``ibmmq.CMQC`` that starts with
``MQOO_``.
+
+ :param open_options: The integer bitmask used when opening an MQ queue
+ (e.g., ``MQOO_INPUT_EXCLUSIVE |
MQOO_FAIL_IF_QUIESCING``).
+
+ :return: A list of the names of the MQ open flags that are set in the
bitmask.
+ For example, ``['MQOO_INPUT_EXCLUSIVE',
'MQOO_FAIL_IF_QUIESCING']``.
+
+ Example:
+ >>> open_options = ibmmq.CMQC.MQOO_INPUT_SHARED |
ibmmq.CMQC.MQOO_FAIL_IF_QUIESCING
+ >>> cls.get_open_options_flags(open_options)
+ ['MQOO_INPUT_SHARED', 'MQOO_FAIL_IF_QUIESCING']
+ """
+ import ibmmq
+
+ return [
Review Comment:
This module imports `ibmmq` inside several methods. If `ibmmq` is meant to
be an optional extra, these imports should fail with a clear, user-facing
`AirflowOptionalProviderFeatureException` (or similar) rather than a raw
`ImportError`, and it’s worth centralizing the import into a single helper to
avoid repeating it throughout the hook.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
