Added support for Easy Connect strings found in tnsnames.ora files.

Author: anthony-tuininga

doc/src/release_notes.rst

@@ -16,6 +16,7 @@ Thin Mode Changes
 #)  Added support for Oracle Database 23c feature that can improve the
     performance of connection creation by reducing the number of round trips
     required for all connections created.
+#)  Added support for Easy Connect strings found in tnsnames.ora files.
 #)  Added support for writing UTF-8 encoded bytes to CLOB and NCLOB values and
     writing strings to BLOB values in order to be consistent with what is done
     for string variables.

src/oracledb/base_impl.pxd

@@ -414,6 +414,8 @@ cdef class ConnectParamsImpl:
     cdef object _get_token_expires(self, str token)
     cdef str _get_wallet_password(self)
     cdef int _parse_connect_string(self, str connect_string) except -1
+    cdef int _parse_easy_connect_string(self, str connect_string,
+                                        object match) except -1
     cdef int _process_connect_descriptor(self, dict args) except -1
     cdef int _set_access_token(self, object val, int error_num) except -1
     cdef int _set_access_token_param(self, object val) except -1

src/oracledb/impl/base/connect_params.pyx

@@ -350,32 +350,7 @@ cdef class ConnectParamsImpl:
         # otherwise, see if the connect string is an EasyConnect string
         m = re.search(EASY_CONNECT_PATTERN, connect_string)
         if m is not None:
-
-            # build up arguments
-            args = m.groupdict()
-            connect_string = connect_string[m.end():]
-            params_pos = connect_string.find("?")
-            if params_pos >= 0:
-                params = connect_string[params_pos + 1:]
-                for part in params.split("&"):
-                    name, value = [s.strip() for s in part.split("=", 1)]
-                    name = name.lower()
-                    name = ALTERNATIVE_PARAM_NAMES.get(name, name)
-                    if value.startswith('"') and value.endswith('"'):
-                        value = value[1:-1]
-                    args[name] = value
-
-            # create description list
-            address = self._default_address.copy()
-            address.set_from_args(args)
-            description = self._default_description.copy()
-            description.set_from_connect_data_args(args)
-            description.set_from_description_args(args)
-            description.set_from_security_args(args)
-            description.children = [AddressList()]
-            description.children[0].children.append(address)
-            self.description_list = DescriptionList()
-            self.description_list.children.append(description)
+            self._parse_easy_connect_string(connect_string, m)
 
         # otherwise, see if the name is a connect alias in a tnsnames.ora
         # configuration file
@@ -386,8 +361,50 @@ cdef class ConnectParamsImpl:
             if connect_string is None:
                 errors._raise_err(errors.ERR_TNS_ENTRY_NOT_FOUND, name=name,
                                   file_name=tnsnames_file.file_name)
-            _parse_connect_descriptor(connect_string, args)
-            self._process_connect_descriptor(args)
+            m = re.search(EASY_CONNECT_PATTERN, connect_string)
+            if m is not None:
+                self._parse_easy_connect_string(connect_string, m)
+            else:
+                _parse_connect_descriptor(connect_string, args)
+                self._process_connect_descriptor(args)
+
+    cdef int _parse_easy_connect_string(self, str connect_string,
+                                        object match) except -1:
+        """
+        Internal method for parsing an Easy Connect string.
+        """
+        cdef:
+            str params, part, name, value, s
+            Description description
+            ssize_t params_pos
+            Address address
+            dict args
+
+        # determine arguments
+        args = match.groupdict()
+        connect_string = connect_string[match.end():]
+        params_pos = connect_string.find("?")
+        if params_pos >= 0:
+            params = connect_string[params_pos + 1:]
+            for part in params.split("&"):
+                name, value = [s.strip() for s in part.split("=", 1)]
+                name = name.lower()
+                name = ALTERNATIVE_PARAM_NAMES.get(name, name)
+                if value.startswith('"') and value.endswith('"'):
+                    value = value[1:-1]
+                args[name] = value
+
+        # create description list
+        address = self._default_address.copy()
+        address.set_from_args(args)
+        description = self._default_description.copy()
+        description.set_from_connect_data_args(args)
+        description.set_from_description_args(args)
+        description.set_from_security_args(args)
+        description.children = [AddressList()]
+        description.children[0].children.append(address)
+        self.description_list = DescriptionList()
+        self.description_list.children.append(description)
 
     cdef int _process_connect_descriptor(self, dict args) except -1:
         """

tests/test_4500_connect_params.py

@@ -776,6 +776,38 @@ def test_4567(self):
         params.set(cclass="")
         self.assertEqual(params.cclass, None)
 
+    def test_4568(self):
+        "4568 - test easy connect string with protocol specified"
+        protocol = "tcp"
+        host = "my_host_4568"
+        port = 1668
+        service_name = "my_service_4568"
+        connect_string = f"{protocol}://{host}:{port}/{service_name}"
+        params = oracledb.ConnectParams()
+        params.parse_connect_string(connect_string)
+        self.assertEqual(params.protocol, protocol)
+        self.assertEqual(params.host, host)
+        self.assertEqual(params.port, port)
+        self.assertEqual(params.service_name, service_name)
+
+    def test_4569(self):
+        "4569 - test easy connect string in tnsnames.ora"
+        alias_name = "tns_alias_4569"
+        host = "my_host4569"
+        port = 1568
+        service_name = "my_service_name_4569"
+        connect_string = f"tcp://{host}:{port}/{service_name}"
+        alias = f"{alias_name} = {connect_string}"
+        with tempfile.TemporaryDirectory() as temp_dir:
+            file_name = os.path.join(temp_dir, "tnsnames.ora")
+            with open(file_name, "w") as f:
+                f.write(alias)
+            params = oracledb.ConnectParams(config_dir=temp_dir)
+            params.parse_connect_string(alias_name)
+        self.assertEqual(params.host, host)
+        self.assertEqual(params.port, port)
+        self.assertEqual(params.service_name, service_name)
+
 
 if __name__ == "__main__":
     test_env.run_test_cases()