Fixed bug resulting in explicit request boundaries to aid planned

database maintenance not being sent when using connection pools with
asyncio; pooled connections that are no longer needed are now closed
normally if possible instead of simply having the socket disconnected
(#393).

Author: anthony-tuininga

doc/src/release_notes.rst

@@ -25,6 +25,11 @@ Thin Mode Changes
 #)  Fixed bug when connecting with asyncio using the parameter ``https_proxy``.
 #)  Fixed regression when connecting where only the host specified by the
     ``https_proxy`` parameter can successfully perform name resolution.
+#)  Fixed bug resulting in explicit request boundaries to aid planned database
+    maintenance not being sent when using connection pools with asyncio.
+#)  Pooled connections that are no longer needed are now closed normally if
+    possible instead of simply having the socket disconnected
+    (`issue 393 <https://github.com/oracle/python-oracledb/issues/393>`__).
 
 Thick Mode Changes
 ++++++++++++++++++

src/oracledb/base_impl.pxd

@@ -586,6 +586,7 @@ cdef class ConnectParamsImpl:
     cdef str _get_token(self)
     cdef object _get_public_instance(self)
     cdef object _get_token_expires(self, str token)
+    cdef bint _get_uses_drcp(self)
     cdef str _get_wallet_password(self)
     cdef int _parse_connect_string(self, str connect_string) except -1
     cdef int _set_access_token(self, object val, int error_num) except -1

src/oracledb/connection.py

@@ -610,6 +610,7 @@ def __init__(
         """
 
         super().__init__()
+        self._pool = pool
 
         # determine if thin mode is being used
         with driver_mode.get_manager() as mode_mgr:
@@ -664,17 +665,29 @@ def __init__(
 
     def __del__(self):
         if self._impl is not None:
-            self._impl.close(in_del=True)
-            self._impl = None
+            self._close(in_del=True)
 
     def __enter__(self):
         self._verify_connected()
         return self
 
     def __exit__(self, exc_type, exc_value, exc_tb):
         if self._impl is not None:
-            self._impl.close(in_del=True)
-            self._impl = None
+            self._close()
+
+    def _close(self, in_del=False):
+        """
+        Closes the connection and makes it unusable for further operations. An
+        Error exception will be raised if any operation is attempted with this
+        connection after this method completes successfully.
+        """
+        if self._pool is not None:
+            pool_impl = self._pool._impl
+            if pool_impl is not None:
+                pool_impl.return_connection(self._impl, in_del)
+        else:
+            self._impl.close(in_del)
+        self._impl = None
 
     def _create_queue(self, impl):
         """
@@ -743,8 +756,7 @@ def close(self) -> None:
         connection after this method completes successfully.
         """
         self._verify_connected()
-        self._impl.close()
-        self._impl = None
+        self._close()
 
     def commit(self) -> None:
         """
@@ -1574,6 +1586,7 @@ def __init__(
         directly but only indirectly through async_connect().
         """
         super().__init__()
+        self._pool = pool
         self._connect_coroutine = self._connect(dsn, pool, params, kwargs)
 
     def __await__(self):
@@ -1590,8 +1603,21 @@ async def __aenter__(self):
 
     async def __aexit__(self, *exc_info):
         if self._impl is not None:
-            await self._impl.close()
-            self._impl = None
+            await self._close()
+
+    async def _close(self, in_del=False):
+        """
+        Closes the connection and makes it unusable for further operations. An
+        Error exception will be raised if any operation is attempted with this
+        connection after this method completes successfully.
+        """
+        if self._pool is not None:
+            pool_impl = self._pool._impl
+            if pool_impl is not None:
+                await pool_impl.return_connection(self._impl, in_del)
+        else:
+            await self._impl.close(in_del)
+        self._impl = None
 
     async def _connect(self, dsn, pool, params, kwargs):
         """
@@ -1718,8 +1744,7 @@ async def close(self) -> None:
         Closes the connection.
         """
         self._verify_connected()
-        await self._impl.close()
-        self._impl = None
+        await self._close()
 
     async def commit(self) -> None:
         """

src/oracledb/impl/base/connect_params.pyx

@@ -294,6 +294,17 @@ cdef class ConnectParamsImpl:
         header = json.loads(base64.b64decode(header_seg))
         return datetime.datetime.utcfromtimestamp(header["exp"])
 
+    cdef bint _get_uses_drcp(self):
+        """
+        Returns a boolean indicating if any of the descriptions associated with
+        the parameters make use of DRCP.
+        """
+        cdef Description description
+        for description in self.description_list.children:
+            if description.server_type == "pooled":
+                return True
+        return False
+
     cdef str _get_wallet_password(self):
         """
         Returns the wallet password, after removing the obfuscation.

src/oracledb/impl/base/pool.pyx

@@ -89,6 +89,9 @@ cdef class BasePoolImpl:
     def reconfigure(self, uint32_t min, uint32_t max, uint32_t increment):
         errors._raise_not_supported("reconfiguring a pool")
 
+    def return_connection(self, BaseConnImpl conn_impl, in_del=False):
+        errors._raise_not_supported("returning a connection to a pool")
+
     def set_getmode(self, uint8_t value):
         errors._raise_not_supported("setting the 'get' mode of a pool")
 

src/oracledb/impl/thick/pool.pyx

@@ -320,6 +320,12 @@ cdef class ThickPoolImpl(BasePoolImpl):
         self.max = max
         self.increment = increment
 
+    def return_connection(self, ThickConnImpl conn_impl, bint in_del=False):
+        """
+        Internal method for returning a connection to the pool.
+        """
+        conn_impl.close(in_del)
+
     def set_getmode(self, uint8_t value):
         """
         Internal method for setting the method by which connections are

src/oracledb/impl/thin/connection.pyx

@@ -46,7 +46,6 @@ cdef class BaseThinConnImpl(BaseConnImpl):
         bint _client_identifier_modified
         str _module
         bint _module_modified
-        BaseThinPoolImpl _pool
         bytes _ltxid
         str _current_schema
         bint _current_schema_modified
@@ -70,6 +69,7 @@ cdef class BaseThinConnImpl(BaseConnImpl):
         int _dbobject_type_cache_num
         bytes _combo_key
         str _connection_id
+        bint _is_pooled
         bint _is_pool_extra
         bytes _transaction_context
         uint8_t pipeline_mode
@@ -174,11 +174,6 @@ cdef class BaseThinConnImpl(BaseConnImpl):
         message.context = self._transaction_context
         return message
 
-    cdef int _force_close(self) except -1:
-        self._pool = None
-        self._clear_dbobject_type_cache()
-        self._protocol._force_close()
-
     cdef Statement _get_statement(self, str sql = None,
                                   bint cache_statement = False):
         """
@@ -198,11 +193,8 @@ cdef class BaseThinConnImpl(BaseConnImpl):
         self._drcp_enabled = description.server_type == "pooled"
         if self._cclass is None:
             self._cclass = description.cclass
-        if self._cclass is None and self._pool is not None \
-                and self._drcp_enabled:
-            gen_uuid = uuid.uuid4()
-            self._cclass = f"DPY:{base64.b64encode(gen_uuid.bytes).decode()}"
-            params._default_description.cclass = self._cclass
+        if self._cclass is None:
+            self._cclass = params._default_description.cclass
 
     cdef int _post_connect_phase_two(self, ConnectParamsImpl params) except -1:
         """
@@ -350,6 +342,13 @@ cdef class ThinConnImpl(BaseThinConnImpl):
         BaseThinConnImpl.__init__(self, dsn, params)
         self._protocol = Protocol()
 
+    cdef int _close(self):
+        """
+        Internal method for closing the connection.
+        """
+        cdef Protocol protocol = <Protocol> self._protocol
+        protocol._close(self)
+
     cdef int _connect_with_address(self, Address address,
                                    Description description,
                                    ConnectParamsImpl params,
@@ -441,9 +440,12 @@ cdef class ThinConnImpl(BaseThinConnImpl):
         protocol._process_single_message(message)
 
     def close(self, bint in_del=False):
+        """
+        Internal method for closing the connection to the database.
+        """
         cdef Protocol protocol = <Protocol> self._protocol
         try:
-            protocol._close(self)
+            protocol.close(self, in_del)
         except (ssl.SSLError, exceptions.DatabaseError):
             pass
 
@@ -455,17 +457,17 @@ cdef class ThinConnImpl(BaseThinConnImpl):
         protocol._process_single_message(message)
 
     def connect(self, ConnectParamsImpl params):
-        # specify that binding a string to a LOB value is possible in thin
-        # mode without the use of asyncio (will be removed in a future release)
-        self._allow_bind_str_to_lob = True
-
+        cdef Protocol protocol = <Protocol> self._protocol
         try:
             self._pre_connect(params)
             self._connect_with_params(params)
             self._post_connect_phase_two(params)
         except:
-            self._force_close()
+            protocol._disconnect()
             raise
+        # specify that binding a string to a LOB value is possible in thin
+        # mode without the use of asyncio (will be removed in a future release)
+        self._allow_bind_str_to_lob = True
 
     def create_queue_impl(self):
         return ThinQueueImpl.__new__(ThinQueueImpl)
@@ -954,7 +956,7 @@ cdef class AsyncThinConnImpl(BaseThinConnImpl):
         """
         cdef BaseAsyncProtocol protocol = <BaseAsyncProtocol> self._protocol
         try:
-            await protocol._close(self)
+            await protocol.close(self, in_del)
         except (ssl.SSLError, exceptions.DatabaseError):
             pass
 
@@ -979,7 +981,7 @@ cdef class AsyncThinConnImpl(BaseThinConnImpl):
             await self._connect_with_params(params)
             self._post_connect_phase_two(params)
         except:
-            self._force_close()
+            protocol._disconnect()
             raise
 
     def create_queue_impl(self):

src/oracledb/impl/thin/messages/auth.pyx

@@ -271,7 +271,7 @@ cdef class AuthMessage(Message):
         # to a pool
         if description.purity == PURITY_DEFAULT \
                 and self.conn_impl._drcp_enabled:
-            if self.conn_impl._pool is None:
+            if self.conn_impl._is_pooled:
                 self.purity = PURITY_NEW
             else:
                 self.purity = PURITY_SELF

src/oracledb/impl/thin/messages/base.pyx

@@ -72,7 +72,7 @@ cdef class Message:
                                   code=self.error_info.num,
                                   offset=self.error_info.pos)
             if error.is_session_dead:
-                self.conn_impl._protocol._force_close()
+                self.conn_impl._protocol._disconnect()
             raise error.exc_type(error)
 
     cdef int _initialize(self, BaseThinConnImpl conn_impl) except -1: