airflow.providers.ibm.mq.hooks.mq¶

Attributes¶

ibmmq

Exceptions¶

IBMMQError

Lightweight wrapper for IBM MQ errors raised by the consumer thread.

Classes¶

`IBMMQConsumer`	Thread worker that consumes one message from an IBM MQ queue.
`IBMMQHook`	Interact with IBM MQ queue managers to consume and produce messages.

Functions¶

requires_ibmmq(func)

Ensure the optional ibmmq module is available.

Module Contents¶

airflow.providers.ibm.mq.hooks.mq.ibmmq = None[source]¶

airflow.providers.ibm.mq.hooks.mq.requires_ibmmq(func)[source]¶

Ensure the optional ibmmq module is available.

Use this decorator on functions or methods that call into the native ibmmq extension so callers receive a clear AirflowOptionalProviderFeatureException when the dependency is not installed.

exception airflow.providers.ibm.mq.hooks.mq.IBMMQError(reason, comp, transient, message)[source]¶

Bases: Exception

Lightweight wrapper for IBM MQ errors raised by the consumer thread.

This allows the async event-loop code in IBMMQHook.aconsume() to handle MQ errors without importing the heavy ibmmq C extension on the event-loop thread.

Parameters:

reason (int) – The integer MQ reason code (e.g. MQRC_CONNECTION_BROKEN).
comp (int) – The integer MQ completion code.
transient (bool) – Whether this error is considered transient (eligible for retry).
message (str) – Human-readable description of the error.

Note:

When reason/comp equal _NON_MQ_SENTINEL (-1), the error was not produced by an ibmmq.MQMIError and no MQ codes are available (for example ibmmq.PYIFError or a wrapped ConnectionError).

reason[source]¶

comp[source]¶

transient[source]¶

class airflow.providers.ibm.mq.hooks.mq.IBMMQConsumer(hook, connection, queue_name, poll_interval, loop, future, stop_event)[source]¶

Bases: threading.Thread, airflow.utils.log.logging_mixin.LoggingMixin

Thread worker that consumes one message from an IBM MQ queue.

The consumer is used by IBMMQHook.aconsume() to execute blocking MQ calls in a dedicated thread because the IBM MQ C client requires handles to be used from the thread that created them. The result (or exception) is forwarded to the asyncio event loop through future.

Parameters:

hook (IBMMQHook) – Hook instance that provides connection and queue helpers.
connection (airflow.providers.common.compat.sdk.Connection) – Airflow connection object resolved from hook.conn_id.
queue_name (str) – Queue to consume from.
poll_interval (float) – Maximum wait time (seconds) for each q.get call.
loop (asyncio.AbstractEventLoop) – Event loop that owns future.
future (asyncio.Future) – Future completed with the decoded message or an exception.
stop_event (threading.Event) – Signal used to stop polling after cancellation.

hook[source]¶

connection[source]¶

queue_name[source]¶

poll_interval[source]¶

loop[source]¶

future[source]¶

stop_event[source]¶

consume(queue_name, poll_interval, stop_event)[source]¶

Blocking implementation that consumes a single message from the given IBM MQ queue.

All IBM MQ handles (queue manager connection, queue) are created and used within this method, satisfying the thread-affinity requirement of the IBM MQ C client library. The ‘stop_event’ is checked between ‘q.get’ calls so the thread terminates promptly after the coroutine side is canceled. Reads are performed with MQGMO_NO_SYNCPOINT, so this method provides at-most-once delivery: once q.get returns successfully, IBM MQ has already committed the message removal from the queue.

MQ-specific exceptions are caught and re-raised as IBMMQError so that the async caller never needs to import the heavy ibmmq C extension.

For an asynchronous interface see IBMMQHook.aconsume().

run()[source]¶

Method representing the thread’s activity.

You may override this method in a subclass. The standard run() method invokes the callable object passed to the object’s constructor as the target argument, if any, with sequential and keyword arguments taken from the args and kwargs arguments, respectively.

class airflow.providers.ibm.mq.hooks.mq.IBMMQHook(conn_id=default_conn_name, queue_manager=None, channel=None, open_options=None)[source]¶

Bases: airflow.providers.common.compat.sdk.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.

The asynchronous consume path intentionally uses MQGMO_NO_SYNCPOINT. That keeps reads non-transactional and therefore provides at-most-once delivery semantics for trigger-based consumption: once q.get returns, IBM MQ has already removed the message from the queue. If the coroutine is canceled after that point but before Airflow yields a TriggerEvent, the message can be lost. The stop-event machinery only prevents additional q.get calls after cancellation; it cannot make the non-transactional get atomic with TriggerEvent emission.

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.

Parameters:

conn_id (str) – Airflow connection ID for the IBM MQ instance. Defaults to "mq_default".
queue_manager (str | None) – 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.
channel (str | None) – MQ channel name used for the connection. If not provided, the value is read from the channel key in the connection’s extra JSON.
open_options (int | None) – 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'[source]¶

default_conn_name = 'mq_default'[source]¶

conn_type = 'ibmmq'[source]¶

hook_name = 'IBM MQ'[source]¶

default_open_options = 'MQOO_INPUT_SHARED'[source]¶

conn_id = 'mq_default'[source]¶

queue_manager = None[source]¶

channel = None[source]¶

open_options = None[source]¶

classmethod parse_open_options(value)[source]¶

Parse MQ open-options from allowed formats into an integer bitmask.

Accepts: - int (returned as-is) - numeric string (decimal or hex, e.g., “2” or “0x10”) - single symbol name from ibmmq.CMQC (e.g., “MQOO_INPUT_SHARED”) - pipe- or comma-separated symbols (e.g. “MQOO_INPUT_SHARED | MQOO_FAIL_IF_QUIESCING”)

Raises ValueError on unknown symbol tokens and TypeError for unsupported types.

classmethod get_ui_field_behaviour()[source]¶

Return custom UI field behaviour for IBM MQ Connection.

classmethod get_open_options_flags(open_options)[source]¶

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_.

Parameters:: open_options (int) – The integer bitmask used when opening an MQ queue (e.g., MQOO_INPUT_EXCLUSIVE | MQOO_FAIL_IF_QUIESCING).
Returns:: 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'].
Return type:: list[str]

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']

get_open_options(queue_name)[source]¶

get_conn(connection=None)[source]¶

Sync context manager for IBM MQ connection lifecycle.

Must be called from the executor thread (not the event loop thread). Retrieves the Airflow connection (or uses the explicitly supplied one), extracts MQ parameters, and manages the IBM MQ connection lifecycle.

Parameters:: connection (airflow.providers.common.compat.sdk.Connection | None) – Optional Airflow connection object. When omitted, the connection is resolved from self.conn_id.
Yield:: IBM MQ connection object

async aconsume(queue_name, poll_interval=5)[source]¶

Asynchronous version of consume().

Wait for a single message from the specified IBM MQ queue and return its decoded payload.

The method retries with exponential back-off whenever the underlying consume() returns None (connection broken, timeout) or raises an unexpected exception, so that an AssetWatcher is never silently killed by a transient failure.

All blocking IBM MQ operations (‘connect’, ‘open’, ‘get’, ‘close’, ‘disconnect’) run in a separate thread via ‘sync_to_async’ to satisfy the IBM MQ C client’s thread-affinity requirement — every operation on a connection must happen from the thread that created it.

A threading.Event stop signal is passed to the worker so that, when this coroutine is canceled (e.g. because the Airflow triggerer reassigns the watcher to another pod), the background thread exits cleanly after the current ‘q.get’ call times out (at most ‘poll_interval’ seconds). This prevents orphaned threads from continuing to poll after cancellation, but it does not change the delivery guarantee: consume() uses MQGMO_NO_SYNCPOINT, so cancellation after q.get returns but before the trigger yields an event can still lose that message.

Parameters:

queue_name (str) – Name of the IBM MQ queue to consume messages from.
poll_interval (float) – Interval in seconds used to wait for messages and to control how long the underlying MQ ‘get’ operation blocks before checking again.

Returns:

The decoded message payload.

Return type:

str | None

async aproduce(queue_name, payload, open_options=None)[source]¶

Asynchronous version of produce().

Put a message onto the specified IBM MQ queue.

All blocking IBM MQ operations run in a separate thread via ‘sync_to_async’ for the same thread-safety reasons as aconsume().

Parameters:

queue_name (str) – Name of the IBM MQ queue to which the message should be sent.
payload (str) – Message payload to send. The payload will be encoded as UTF-8 before being placed on the queue.
open_options (int | None) – Integer bitmask of MQOO_* open options for the queue. If not provided, defaults to MQOO_OUTPUT.

Returns:

None

Return type:

None

produce(connection, queue_name, payload, open_options=None)[source]¶

Blocking implementation of aproduce().

Parameters:

connection (airflow.providers.common.compat.sdk.Connection) – Airflow connection object.
queue_name (str) – Name of the IBM MQ queue to which the message should be sent.
payload (str) – Message payload to send. The payload will be encoded as UTF-8 before being placed on the queue.
open_options (int | None) – Integer bitmask of MQOO_* open options for the queue. If not provided, defaults to MQOO_OUTPUT.