Improved detection of the signature used by output type handlers, in

particular those that that make use of __call__().

Author: anthony-tuininga

doc/src/release_notes.rst

@@ -46,6 +46,8 @@ Common Changes
 #)  Added support for maintainers to specify optional compilation arguments
     when building python-oracledb. A new environment variable
     ``PYO_COMPILE_ARGS`` can be set :ref:`before building <installsrc>`.
+#)  Improved detection of the signature used by output type handlers, in
+    particular those that that make use of ``__call__()``.
 #)  Python wheel package binaries for Linux on `PyPI
     <https://pypi.org/project/oracledb/>`__ are now stripped to reduce their
     size.

src/oracledb/impl/base/cursor.pyx

@@ -316,11 +316,11 @@ cdef class BaseCursorImpl:
             conn_impl = self._get_conn_impl()
             type_handler = conn_impl.outputtypehandler
         if type_handler is not None:
-            uses_fetch_info[0] = \
-                    (inspect.isfunction(type_handler) and \
-                    type_handler.__code__.co_argcount == 2) or \
-                    (inspect.ismethod(type_handler) and \
-                    type_handler.__code__.co_argcount == 3)
+            try:
+                sig = inspect.signature(type_handler)
+                uses_fetch_info[0] = (len(sig.parameters) == 2)
+            except (ValueError, TypeError):
+                uses_fetch_info[0] = False
         return type_handler
 
     cdef int _init_fetch_vars(self, uint32_t num_columns) except -1:

tests/test_3800_typehandler.py

@@ -26,6 +26,7 @@
 3800 - Module for testing the input and output type handlers.
 """
 
+import datetime
 import json
 import unittest
 
@@ -286,6 +287,42 @@ def output_type_handler(cursor, metadata):
         expected_result = f"A{rc}B{rc}C{rc}"
         self.assertEqual(result, expected_result)
 
+    def test_3808(self):
+        "3808 - output type handler with object implementing __call__()"
+
+        class TimestampOutputTypeHandler:
+
+            def __init__(self, unit="s"):
+                if unit == "ms":
+                    self.factor = 1000
+                else:
+                    self.factor = 1
+
+            def converter(self, d):
+                return int(d.timestamp() * self.factor)
+
+            def __call__(self, cursor, metadata):
+                if metadata.type_code is oracledb.DB_TYPE_TIMESTAMP:
+                    return cursor.var(
+                        metadata.type_code,
+                        arraysize=cursor.arraysize,
+                        outconverter=self.converter,
+                    )
+
+        d = datetime.datetime.today()
+        with self.conn.cursor() as cursor:
+            cursor.outputtypehandler = TimestampOutputTypeHandler("ms")
+            cursor.setinputsizes(oracledb.DB_TYPE_TIMESTAMP)
+            cursor.execute("select :d from dual", [d])
+            (result,) = cursor.fetchone()
+            self.assertEqual(result, int(d.timestamp() * 1000))
+        with self.conn.cursor() as cursor:
+            cursor.outputtypehandler = TimestampOutputTypeHandler("s")
+            cursor.setinputsizes(oracledb.DB_TYPE_TIMESTAMP)
+            cursor.execute("select :d from dual", [d])
+            (result,) = cursor.fetchone()
+            self.assertEqual(result, int(d.timestamp()))
+
 
 if __name__ == "__main__":
     test_env.run_test_cases()

tests/test_6000_typehandler_async.py

@@ -26,6 +26,7 @@
 6000 - Module for testing the input and output type handlers with asyncio.
 """
 
+import datetime
 import json
 import unittest
 
@@ -258,6 +259,42 @@ def output_type_handler(cursor, metadata):
         await self.cursor.execute("select * from TestJson")
         self.assertEqual(await self.cursor.fetchall(), data_to_insert)
 
+    async def test_6007(self):
+        "6007 - output type handler with object implementing __call__()"
+
+        class TimestampOutputTypeHandler:
+
+            def __init__(self, unit="s"):
+                if unit == "ms":
+                    self.factor = 1000
+                else:
+                    self.factor = 1
+
+            def converter(self, d):
+                return int(d.timestamp() * self.factor)
+
+            def __call__(self, cursor, metadata):
+                if metadata.type_code is oracledb.DB_TYPE_TIMESTAMP:
+                    return cursor.var(
+                        metadata.type_code,
+                        arraysize=cursor.arraysize,
+                        outconverter=self.converter,
+                    )
+
+        d = datetime.datetime.today()
+        with self.conn.cursor() as cursor:
+            cursor.outputtypehandler = TimestampOutputTypeHandler("ms")
+            cursor.setinputsizes(oracledb.DB_TYPE_TIMESTAMP)
+            await cursor.execute("select :d from dual", [d])
+            (result,) = await cursor.fetchone()
+            self.assertEqual(result, int(d.timestamp() * 1000))
+        with self.conn.cursor() as cursor:
+            cursor.outputtypehandler = TimestampOutputTypeHandler("s")
+            cursor.setinputsizes(oracledb.DB_TYPE_TIMESTAMP)
+            await cursor.execute("select :d from dual", [d])
+            (result,) = await cursor.fetchone()
+            self.assertEqual(result, int(d.timestamp()))
+
 
 if __name__ == "__main__":
     test_env.run_test_cases()