Added support for array enqueue and dequeue for AQ in thin mode.

Author: anthony-tuininga

doc/src/api_manual/aq.rst

@@ -21,7 +21,7 @@ used to enqueue and dequeue messages.
 Queue Methods
 -------------
 
-.. method:: Queue.deqmany(maxMessages)
+.. method:: Queue.deqmany(max_num_messages)
 
     Dequeues up to the specified number of messages from the queue and returns
     a list of these messages. Each element of the returned list is a
@@ -50,12 +50,14 @@ Queue Methods
 
     .. warning::
 
-        Prior to Oracle Database 21c, calling this function in parallel on
-        different connections acquired from the same pool may fail due to
-        Oracle bug 29928074. Either ensure that this function is not run in
-        parallel, use standalone connections or connections from different
-        pools, or make multiple calls to :meth:`Queue.enqone()` instead. The
-        function :meth:`Queue.deqmany()` call is not affected.
+        In python-oracledb Thick mode using Oracle Client libraries prior to
+        21c, calling :meth:`Queue.enqmany()` in parallel on different
+        connections acquired from the same connection pool may fail due to
+        Oracle bug 29928074. To avoid this, do one of: upgrade the client
+        libraries, ensure that :meth:`Queue.enqmany()` is not run in parallel,
+        use standalone connections or connections from different pools, or make
+        multiple calls to :meth:`Queue.enqone()`. The function
+        :meth:`Queue.deqmany()` call is not affected.
 
     For consistency and compliance with the PEP 8 naming style, the name of
     the method was changed from `enqMany()`. The old name will continue

doc/src/api_manual/async_aq.rst

@@ -23,12 +23,25 @@ are used to enqueue and dequeue messages.
 AsyncQueue Methods
 ------------------
 
+.. method:: AsyncQueue.deqmany(max_num_messages)
+
+    Dequeues up to the specified number of messages from the queue and returns
+    a list of these messages. Each element of the returned list is a
+    :ref:`message property <msgproperties>` object.
+
 .. method:: AsyncQueue.deqone()
 
     Dequeues at most one message from the queue. If a message is dequeued, it
     will be a :ref:`message property <asyncmsgproperties>` object; otherwise,
     the value *None* will be returned.
 
+.. method:: AsyncQueue.enqmany(messages)
+
+    Enqueues multiple messages into the queue. The ``messages`` parameter must
+    be a sequence containing :ref:`message property <msgproperties>` objects
+    which have all had their payload attribute set to a value that the queue
+    supports.
+
 .. method:: AsyncQueue.enqone(message)
 
     Enqueues a single message into the queue. The message must be a

doc/src/release_notes.rst

@@ -17,10 +17,12 @@ oracledb 3.1.0 (TBD)
 Thin Mode Changes
 +++++++++++++++++
 
-#)  Added :ref:`Oracle Advanced Queuing <aqusermanual>` support for single
-    enqueue and dequeue of JSON payloads.
-#)  Added Async :ref:`Oracle Advanced Queuing <asyncaq>` support for single
-    enqueue and dequeue of RAW and Oracle object payload types.
+#)  Improved support for :ref:`Oracle Advanced Queuing <aqusermanual>`:
+
+    - added support for JSON payloads
+    - added support for bulk enqueuing and dequeuing
+    - added support for using AQ with asyncio
+
 #)  Improved error message when the cryptography package cannot be imported
     (`issue 455 <https://github.com/oracle/python-oracledb/issues/455>`__).
 #)  Fixed decoding of nested PL/SQL records

doc/src/user_guide/aq.rst

@@ -26,9 +26,8 @@ types.
 - JSON payloads require Oracle Database 21c (or later). In python-oracle Thick
   mode, Oracle Client libraries 21c (or later) are also needed.
 
-JMS payloads, array message queuing and dequeuing operations, and
-:ref:`Recipient Lists <reciplists>` are only supported in python-oracledb
-:ref:`Thick mode <enablingthick>`.
+JMS payloads and :ref:`Recipient Lists <reciplists>` are only supported in
+python-oracledb :ref:`Thick mode <enablingthick>`.
 
 There are examples of AQ Classic Queues in the `GitHub samples
 <https://github.com/oracle/python-oracledb/tree/main/samples>`__ directory.
@@ -343,8 +342,9 @@ message will be dropped from the queue.
 Bulk Enqueue and Dequeue
 ========================
 
-The :meth:`~Queue.enqmany()` and :meth:`~Queue.deqmany()` methods can be used
-for efficient bulk message handling.
+The :meth:`Queue.enqmany()`, :meth:`Queue.deqmany()`,
+:meth:`AsyncQueue.enqmany()`, and :meth:`AsyncQueue.deqmany()` methods can be
+used for efficient bulk message handling.
 
 The :meth:`~Queue.enqmany()` method is similar to :meth:`~Queue.enqone()` but
 accepts an array of messages:
@@ -362,16 +362,18 @@ accepts an array of messages:
 
 .. warning::
 
-    Calling :meth:`~Queue.enqmany()` in parallel on different connections
-    acquired from the same pool may fail due to Oracle bug 29928074. To avoid
-    this, ensure that :meth:`~Queue.enqmany()` is not run in parallel, use
-    standalone connections or connections from different pools, or make
-    multiple calls to :meth:`~Queue.enqone()` instead. The function
-    :meth:`~Queue.deqmany()` call is not affected.
-
-To dequeue multiple messages at one time, use :meth:`~Queue.deqmany()`.  This
-takes an argument specifying the maximum number of messages to dequeue at one
-time:
+    In python-oracledb Thick mode using Oracle Client libraries prior to 21c,
+    calling :meth:`Queue.enqmany()` in parallel on different connections
+    acquired from the same connection pool may fail due to Oracle
+    bug 29928074. To avoid this, do one of: upgrade the client libraries,
+    ensure that :meth:`Queue.enqmany()` is not run in parallel, use standalone
+    connections or connections from different pools, or make multiple calls to
+    :meth:`Queue.enqone()`. The function :meth:`Queue.deqmany()` call is not
+    affected.
+
+To dequeue multiple messages at one time, use :meth:`Queue.deqmany()` or
+:meth:`AsyncQueue.deqmany()`.  This takes an argument specifying the maximum
+number of messages to dequeue at one time:
 
 .. code-block:: python
 

samples/bulk_aq.py

@@ -53,8 +53,9 @@
     "The twelfth and final message",
 ]
 
-# this script is currently only supported in python-oracledb thick mode
-oracledb.init_oracle_client(lib_dir=sample_env.get_oracle_client())
+# determine whether to use python-oracledb thin mode or thick mode
+if not sample_env.get_is_thin():
+    oracledb.init_oracle_client(lib_dir=sample_env.get_oracle_client())
 
 # connect to database
 connection = oracledb.connect(

samples/bulk_aq_async.py

@@ -0,0 +1,107 @@
+# -----------------------------------------------------------------------------
+# Copyright (c) 2025, Oracle and/or its affiliates.
+#
+# Portions Copyright 2007-2015, Anthony Tuininga. All rights reserved.
+#
+# Portions Copyright 2001-2007, Computronix (Canada) Ltd., Edmonton, Alberta,
+# Canada. All rights reserved.
+#
+# This software is dual-licensed to you under the Universal Permissive License
+# (UPL) 1.0 as shown at https://oss.oracle.com/licenses/upl and Apache License
+# 2.0 as shown at http://www.apache.org/licenses/LICENSE-2.0. You may choose
+# either license.
+#
+# If you elect to accept the software under the Apache License, Version 2.0,
+# the following applies:
+#
+# Licensed 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
+#
+#    https://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.
+# -----------------------------------------------------------------------------
+
+# -----------------------------------------------------------------------------
+# bulk_aq_async.py
+#
+# Demonstrates how to use bulk enqueuing and dequeuing of messages with
+# advanced queuing using asyncio. It makes use of a RAW queue created in the
+# sample setup.
+# -----------------------------------------------------------------------------
+
+import asyncio
+
+import oracledb
+import sample_env
+
+QUEUE_NAME = "DEMO_RAW_QUEUE"
+PAYLOAD_DATA = [
+    "The first message",
+    "The second message",
+    "The third message",
+    "The fourth message",
+    "The fifth message",
+    "The sixth message",
+    "The seventh message",
+    "The eighth message",
+    "The ninth message",
+    "The tenth message",
+    "The eleventh message",
+    "The twelfth and final message",
+]
+
+
+async def main():
+
+    # connect to database
+    async with oracledb.connect_async(
+        user=sample_env.get_main_user(),
+        password=sample_env.get_main_password(),
+        dsn=sample_env.get_connect_string(),
+    ) as connection:
+
+        # create a queue
+        queue = connection.queue(QUEUE_NAME)
+        queue.deqoptions.wait = oracledb.DEQ_NO_WAIT
+        queue.deqoptions.navigation = oracledb.DEQ_FIRST_MSG
+
+        # dequeue all existing messages to ensure the queue is empty, just so
+        # that the results are consistent
+        while await queue.deqone():
+            pass
+
+        # enqueue a few messages
+        print("Enqueuing messages...")
+        batch_size = 6
+        data_to_enqueue = PAYLOAD_DATA
+        while data_to_enqueue:
+            batch_data = data_to_enqueue[:batch_size]
+            data_to_enqueue = data_to_enqueue[batch_size:]
+            messages = [
+                connection.msgproperties(payload=d) for d in batch_data
+            ]
+            for data in batch_data:
+                print(data)
+            await queue.enqmany(messages)
+        await connection.commit()
+
+        # dequeue the messages
+        print("\nDequeuing messages...")
+        batch_size = 8
+        while True:
+            messages = await queue.deqmany(batch_size)
+            if not messages:
+                break
+            for props in messages:
+                print(props.payload.decode())
+        await connection.commit()
+        print("\nDone.")
+
+
+asyncio.run(main())

src/oracledb/aq.py

@@ -180,7 +180,7 @@ def enqmany(self, messages: list) -> None:
         acquired from the same pool may fail due to Oracle bug 29928074. Ensure
         that this function is not run in parallel, use standalone connections
         or connections from different pools, or make multiple calls to
-        enqOne() instead. The function Queue.deqMany() call is not affected.
+        enqone() instead. The function Queue.deqmany() call is not affected.
         """
         for message in messages:
             self._verify_message(message)
@@ -211,6 +211,14 @@ def enqOne(self, message: "MessageProperties") -> None:
 
 class AsyncQueue(BaseQueue):
 
+    async def deqmany(self, max_num_messages: int) -> list:
+        """
+        Dequeues up to the specified number of messages from the queue and
+        returns a list of these messages.
+        """
+        message_impls = await self._impl.deq_many(max_num_messages)
+        return [MessageProperties._from_impl(impl) for impl in message_impls]
+
     async def deqone(self) -> Union["MessageProperties", None]:
         """
         Dequeues at most one message from the queue and returns it. If no
@@ -220,6 +228,23 @@ async def deqone(self) -> Union["MessageProperties", None]:
         if message_impl is not None:
             return MessageProperties._from_impl(message_impl)
 
+    async def enqmany(self, messages: list) -> None:
+        """
+        Enqueues multiple messages into the queue. The messages parameter must
+        be a sequence containing message property objects which have all had
+        their payload attribute set to a value that the queue supports.
+
+        Warning: calling this function in parallel on different connections
+        acquired from the same pool may fail due to Oracle bug 29928074. Ensure
+        that this function is not run in parallel, use standalone connections
+        or connections from different pools, or make multiple calls to
+        enqone() instead. The function Queue.deqmany() call is not affected.
+        """
+        for message in messages:
+            self._verify_message(message)
+        message_impls = [m._impl for m in messages]
+        await self._impl.enq_many(message_impls)
+
     async def enqone(self, message: "MessageProperties") -> None:
         """
         Enqueues a single message into the queue. The message must be a message

src/oracledb/impl/thin/constants.pxi

@@ -103,6 +103,11 @@ cdef enum:
 cdef enum:
     TNS_AQ_MSG_NO_DELAY = 0
     TNS_AQ_MSG_NO_EXPIRATION = -1
+    TNS_AQ_ARRAY_ENQ = 0x01
+    TNS_AQ_ARRAY_DEQ = 0x02
+    TNS_AQ_ARRAY_FLAGS_RETURN_MESSAGE_ID = 0x01
+    TNS_TTC_ENQ_STREAMING_ENABLED = 0x00000001
+    TNS_TTC_ENQ_STREAMING_DISABLED = 0x00000000
 
 # AQ flags
 cdef enum:
@@ -340,6 +345,7 @@ cdef enum:
     TNS_FUNC_LOB_OP = 96
     TNS_FUNC_AQ_ENQ = 121
     TNS_FUNC_AQ_DEQ = 122
+    TNS_FUNC_ARRAY_AQ = 145
     TNS_FUNC_LOGOFF = 9
     TNS_FUNC_PING = 147
     TNS_FUNC_PIPELINE_BEGIN = 199