fix(connection): 添加GaussDB兼容的连接参数获取回退方法

- 实现了_get_parameters_fallback方法，支持GaussDB不支持PGconn.info属性时获取连接参数
- 修改get_parameters方法，优先尝试正常路径，失败则调用回退方法
- 调整dsn属性构建函数，增加异常捕获及使用连接参数回退方法生成DSN
- 修正DSN字符串构建逻辑，跳过密码字段并对值进行转义处理
- 添加单元测试验证回退方法的参数获取和DSN生成行为
- 测试连接信息基本属性和编码获取的正确性

Author: 陈云亮

gaussdb/gaussdb/_connection_info.py

@@ -11,7 +11,6 @@
 
 from . import pq
 from ._tz import get_tzinfo
-from .conninfo import make_conninfo
 
 
 class ConnectionInfo:
@@ -72,26 +71,74 @@ def get_parameters(self) -> dict[str, str]:
         either from the connection string and parameters passed to
         `~Connection.connect()` or from environment variables. The password
         is never returned (you can read it using the `password` attribute).
+
+        Note:
+            GaussDB does not support PGconn.info attribute, uses fallback method.
         """
         pyenc = self.encoding
 
-        # Get the known defaults to avoid reporting them
-        defaults = {
-            i.keyword: i.compiled
-            for i in pq.Conninfo.get_defaults()
-            if i.compiled is not None
-        }
-        # Not returned by the libq. Bug? Bet we're using SSH.
-        defaults.setdefault(b"channel_binding", b"prefer")
-        defaults[b"passfile"] = str(Path.home() / ".pgpass").encode()
-
-        return {
-            i.keyword.decode(pyenc): i.val.decode(pyenc)
-            for i in self.pgconn.info
-            if i.val is not None
-            and i.keyword != b"password"
-            and i.val != defaults.get(i.keyword)
-        }
+        # Check if info attribute is supported (GaussDB does not support)
+        try:
+            info = self.pgconn.info
+            if info is None:
+                return self._get_parameters_fallback()
+        except (AttributeError, NotImplementedError):
+            return self._get_parameters_fallback()
+
+        # PostgreSQL normal path
+        try:
+            # Get the known defaults to avoid reporting them
+            defaults = {
+                i.keyword: i.compiled
+                for i in pq.Conninfo.get_defaults()
+                if i.compiled is not None
+            }
+            # Not returned by the libq. Bug? Bet we're using SSH.
+            defaults.setdefault(b"channel_binding", b"prefer")
+            defaults[b"passfile"] = str(Path.home() / ".pgpass").encode()
+
+            return {
+                i.keyword.decode(pyenc): i.val.decode(pyenc)
+                for i in info
+                if i.val is not None
+                and i.keyword != b"password"
+                and i.val != defaults.get(i.keyword)
+            }
+        except Exception:
+            # Use fallback on error
+            return self._get_parameters_fallback()
+
+    def _get_parameters_fallback(self) -> dict[str, str]:
+        """Fallback method for getting connection parameters.
+
+        When PGconn.info is not available (e.g., GaussDB),
+        retrieve basic connection information from other sources.
+        """
+        params = {}
+
+        # Get available information from pgconn attributes
+        if self.pgconn.host:
+            params["host"] = self.pgconn.host.decode(self.encoding, errors="replace")
+
+        if self.pgconn.port:
+            params["port"] = self.pgconn.port.decode(self.encoding, errors="replace")
+
+        if self.pgconn.db:
+            params["dbname"] = self.pgconn.db.decode(self.encoding, errors="replace")
+
+        if self.pgconn.user:
+            params["user"] = self.pgconn.user.decode(self.encoding, errors="replace")
+
+        # Get other available parameters
+        try:
+            if hasattr(self.pgconn, "options") and self.pgconn.options:
+                params["options"] = self.pgconn.options.decode(
+                    self.encoding, errors="replace"
+                )
+        except Exception:
+            pass
+
+        return params
 
     @property
     def dsn(self) -> str:
@@ -103,7 +150,25 @@ def dsn(self) -> str:
         password is never returned (you can read it using the `password`
         attribute).
         """
-        return make_conninfo(**self.get_parameters())
+        try:
+            params = self.get_parameters()
+        except Exception:
+            params = self._get_parameters_fallback()
+
+        if not params:
+            return ""
+
+        # Build DSN string
+        parts = []
+        for key, value in params.items():
+            if key == "password":
+                continue  # Do not include password
+            # Escape values
+            if " " in value or "=" in value or "'" in value:
+                value = "'" + value.replace("'", "\\'") + "'"
+            parts.append(f"{key}={value}")
+
+        return " ".join(parts)
 
     @property
     def status(self) -> pq.ConnStatus:

tests/test_connection_info.py

@@ -262,3 +262,47 @@ def test_set_encoding_unsupported(conn):
 
 def test_vendor(conn):
     assert conn.info.vendor
+
+
+class TestConnectionInfoFallback:
+    """Test GaussDB fallback logic."""
+
+    def test_get_parameters_fallback(self, conn):
+        """Test fallback method for getting parameters."""
+        params = conn.info.get_parameters()
+
+        # Should have at least some basic information
+        # even when using fallback method
+        assert isinstance(params, dict)
+
+        # Verify password is not included
+        assert "password" not in params
+
+    def test_dsn_fallback(self, conn):
+        """Test fallback method for generating DSN."""
+        dsn = conn.info.dsn
+
+        # DSN should be a string
+        assert isinstance(dsn, str)
+
+        # Should not contain password
+        assert "password=" not in dsn.lower()
+
+    def test_basic_info_available(self, conn):
+        """Test basic connection information is available."""
+        info = conn.info
+
+        # These attributes should always be available
+        assert info.host is not None or info.hostaddr is not None
+        assert info.port is not None
+        assert info.dbname is not None
+        assert info.user is not None
+
+    def test_encoding(self, conn):
+        """Test encoding retrieval."""
+        info = conn.info
+
+        # Should be able to get encoding
+        encoding = info.encoding
+        assert encoding is not None
+        assert isinstance(encoding, str)