Added support for enqueuing and dequeing AQ messages as JSON.

Author: anthony-tuininga

doc/src/api_manual/aq.rst

@@ -106,7 +106,8 @@ Queue Attributes
 .. attribute:: Queue.payload_type
 
     This read-only attribute returns the object type for payloads that can be
-    enqueued and dequeued. If using a raw queue, this returns the value None.
+    enqueued and dequeued. If using a JSON queue, this returns the value
+    ``"JSON"``. If using a raw queue, this returns the value ``None``.
 
     For consistency and compliance with the PEP 8 naming style, the name of
     the attribute was changed from `payloadType`. The old name will

doc/src/api_manual/connection.rst

@@ -193,7 +193,8 @@ Connection Methods
 
     The ``payload_type`` parameter, if specified, is expected to be an
     :ref:`object type <dbobjecttype>` that identifies the type of payload the
-    queue expects. If not specified, RAW data is enqueued and dequeued.
+    queue expects. If the string "JSON" is specified, JSON data is enqueued and
+    dequeued. If not specified, RAW data is enqueued and dequeued.
 
     For consistency and compliance with the PEP 8 naming style, the
     parameter `payloadType` was renamed to `payload_type`. The old name

doc/src/release_notes.rst

@@ -31,6 +31,7 @@ Thick Mode Changes
 
 #)  Added support for getting the message id of the AQ message which generated
     a notification.
+#)  Added support for enqueuing and dequeing AQ messages as JSON.
 #)  Added the ability to use `externalauth` as a connection parameter for
     standalone connections in addition to creating pools. For standalone
     connections, this parameter is optional.

doc/src/user_guide/aq.rst

@@ -18,14 +18,22 @@ architecture.
 Python-oracledb uses the updated interface for Oracle Advanced Queuing that
 was first introduced in cx_Oracle 7.2.
 
+Starting from Oracle Database 21c, Advanced Queuing also supports the JSON
+payload type. To use the JSON payload type, the Oracle Client libraries must
+be version 21 or later.
+
 There are Advanced Queuing examples in the `GitHub examples
 <https://github.com/oracle/python-oracledb/tree/main/samples>`__ directory.
 
 
 Creating a Queue
 ================
 
-Before being used, queues need to be created in the database, for example in
+Before being used, queues need to be created in the database.
+
+**Using RAW Payloads**
+
+Queues can be created using the RAW payload type, for example in
 SQL*Plus:
 
 .. code-block:: sql
@@ -37,16 +45,37 @@ SQL*Plus:
     end;
     /
 
-This examples creates a RAW queue suitable for sending string or raw bytes
+This example creates a RAW queue suitable for sending string or bytes
 messages.
 
+**Using JSON Payloads**
+
+Also, queues can be created using the JSON payload type. For example,
+in SQL*Plus:
+
+.. code-block:: sql
+
+    begin
+        dbms_aqadm.create_queue_table('JSON_QUEUE_TABLE', 'JSON');
+        dbms_aqadm.create_queue('DEMO_JSON_QUEUE', 'JSON_QUEUE_TABLE');
+        dbms_aqadm.start_queue('DEMO_JSON_QUEUE');
+    end;
+    /
+
+This example creates a JSON queue suitable for sending JSON data
+messages.
 
 Enqueuing Messages
 ==================
 
 To send messages in Python, you connect and get a :ref:`queue <queue>`. The
 queue can be used for enqueuing, dequeuing, or both as needed.
 
+**Using RAW Payloads**
+
+You can connect to the database and get the queue that was created with RAW
+payload type by using:
+
 .. code-block:: python
 
     queue = connection.queue("DEMO_RAW_QUEUE")
@@ -66,24 +95,63 @@ messages:
     connection.commit()
 
 Since the queue sending the messages is a RAW queue, the strings in this
-example will be internally encoded to bytes using :attr:`Connection.encoding`
+example will be internally encoded to bytes using ``message.encode()``
 before being enqueued.
 
+**Using JSON Payloads**
+
+You can connect to the database and get the queue that was created with JSON
+payload type by using:
+
+.. code-block:: python
+
+    queue = connection.queue("DEMO_JSON_QUEUE", "JSON")
+    # The second argument (JSON) indicates that the queue is of JSON payload type.
+
+Now the message can be enqueued using :meth:`~Queue.enqone()`.
+
+.. code-block:: python
+
+    json_data = [
+        [
+            2.75,
+            True,
+            'Ocean Beach',
+            b'Some bytes',
+            {'keyA': 1.0, 'KeyB': 'Melbourne'},
+            datetime.datetime(2022, 8, 1, 0, 0)
+        ],
+        dict(name="John", age=30, city="New York")
+    ]
+    for data in json_data:
+        queue.enqone(connection.msgproperties(payload=data))
+    connection.commit()
 
 Dequeuing Messages
 ==================
 
 Dequeuing is performed similarly. To dequeue a message call the method
-:meth:`~Queue.deqone()` as shown. Note that if the message is expected to be a
-string, the bytes must be decoded using :attr:`Connection.encoding`.
+:meth:`~Queue.deqone()` as shown in the examples below.
+
+**Using RAW Payload Type**
 
 .. code-block:: python
 
     queue = connection.queue("DEMO_RAW_QUEUE")
-    msg = queue.deqOne()
+    message = queue.deqOne()
     connection.commit()
-    print(msg.payload.decode(connection.encoding))
+    print(message.payload.decode())
+
+Note that if the message is expected to be a string, the bytes must
+be decoded using ``message.payload.decode()``, as shown.
+
+**Using JSON Payload Type**
 
+.. code-block:: python
+
+    queue = connection.queue("DEMO_JSON_QUEUE", "JSON")
+    message = queue.deqOne()
+    connection.commit()
 
 Using Object Queues
 ===================
@@ -133,9 +201,9 @@ Dequeuing is done like this:
     book_type = connection.gettype("UDT_BOOK")
     queue = connection.queue("DEMO_BOOK_QUEUE", book_type)
 
-    msg = queue.deqone()
+    message = queue.deqone()
     connection.commit()
-    print(msg.payload.TITLE)        # will print Quick Brown Fox
+    print(message.payload.TITLE)   # will print Quick Brown Fox
 
 
 Using Recipient Lists
@@ -241,8 +309,8 @@ time:
 
 .. code-block:: python
 
-    for m in queue.deqmany(10):
-        print(m.payload.decode(connection.encoding))
+    for message in queue.deqmany(10):
+        print(message.payload.decode())
 
 Depending on the queue properties and the number of messages available to
 dequeue, this code will print out from zero to ten messages.

src/oracledb/aq.py

@@ -175,8 +175,11 @@ def payload_type(self) -> Union[DbObjectType, None]:
         If using a raw queue, this returns the value None.
         """
         if self._payload_type is None:
-            self._payload_type = \
-                    DbObjectType._from_impl(self._impl.payload_type)
+            if self._impl.is_json:
+                self._payload_type = "JSON"
+            elif self._impl.payload_type is not None:
+                self._payload_type = \
+                        DbObjectType._from_impl(self._impl.payload_type)
         return self._payload_type
 
     @property
@@ -488,16 +491,16 @@ def payload(self) -> Union[bytes, DbObject]:
         return self._impl.payload
 
     @payload.setter
-    def payload(self, value: Union[bytes, DbObject]) -> None:
+    def payload(self, value: object) -> None:
         if isinstance(value, DbObject):
             self._impl.set_payload_object(value._impl)
+        elif not isinstance(value, (str, bytes)):
+            self._impl.set_payload_json(value)
         else:
             if isinstance(value, str):
                 value_bytes = value.encode()
             elif isinstance(value, bytes):
                 value_bytes = value
-            else:
-                raise TypeError("payload must be a DbObject, string or bytes")
             self._impl.set_payload_bytes(value_bytes)
         self._impl.payload = value
 

src/oracledb/base_impl.pxd

@@ -395,6 +395,7 @@ cdef class BaseQueueImpl:
         readonly BaseDbObjectTypeImpl payload_type
         readonly BaseDeqOptionsImpl deq_options_impl
         readonly BaseEnqOptionsImpl enq_options_impl
+        readonly bint is_json
 
 
 cdef class BaseDeqOptionsImpl:

src/oracledb/connection.py

@@ -603,7 +603,7 @@ def prepare(self) -> bool:
         """
         return self.tpc_prepare()
 
-    def queue(self, name: str, payload_type: DbObjectType=None, *,
+    def queue(self, name: str, payload_type: [DbObjectType, str]=None, *,
               payloadType: DbObjectType=None) -> Queue:
         """
         Creates and returns a queue which is used to enqueue and dequeue
@@ -612,24 +612,29 @@ def queue(self, name: str, payload_type: DbObjectType=None, *,
         The name parameter is expected to be a string identifying the queue in
         which messages are to be enqueued or dequeued.
 
-        The payload_type parameter, if specified, is expected to be an object
-        type that identifies the type of payload the queue expects. If not
-        specified, RAW data is enqueued and dequeued.
+        The payload_type parameter, if specified, is expected to be an
+        object type that identifies the type of payload the queue expects.
+        If the string "JSON" is specified, JSON data is enqueued and dequeued.
+        If not specified, RAW data is enqueued and dequeued.
         """
         self._verify_connected()
         payload_type_impl = None
+        is_json = False
         if payloadType is not None:
             if payload_type is not None:
                 errors._raise_err(errors.ERR_DUPLICATED_PARAMETER,
                                   deprecated_name="payloadType",
                                   new_name="payload_type")
             payload_type = payloadType
         if payload_type is not None:
-            if not isinstance(payload_type, DbObjectType):
+            if payload_type == "JSON":
+                is_json = True
+            elif not isinstance(payload_type, DbObjectType):
                 raise TypeError("expecting DbObjectType")
-            payload_type_impl = payload_type._impl
+            else:
+                payload_type_impl = payload_type._impl
         impl = self._impl.create_queue_impl()
-        impl.initialize(self._impl, name, payload_type_impl)
+        impl.initialize(self._impl, name, payload_type_impl, is_json)
         return Queue._from_impl(self, impl)
 
     def rollback(self) -> None:

src/oracledb/impl/base/queue.pyx

@@ -49,7 +49,7 @@ cdef class BaseQueueImpl:
 
     @utils.CheckImpls("initializing a queue")
     def initialize(self, BaseConnImpl conn_impl, str name,
-                   BaseDbObjectImpl payload_type):
+                   BaseDbObjectImpl payload_type, bint is_json):
         pass
 
 

src/oracledb/impl/thick/connection.pyx

@@ -404,6 +404,7 @@ cdef class ThickConnImpl(BaseConnImpl):
     def create_msg_props_impl(self):
         cdef ThickMsgPropsImpl impl
         impl = ThickMsgPropsImpl.__new__(ThickMsgPropsImpl)
+        impl._conn_impl = self
         if dpiConn_newMsgProps(self._handle, &impl._handle) < 0:
             _raise_from_odpi()
         return impl