Added parse_conninfo()

diff --git a/psycopg3/_pq_ctypes.py b/psycopg3/_pq_ctypes.py
index 20078d967cdc163aea750c8552c259ae0af88a41..f014ed27dac0eeb88d9526a76c531cff839df90e 100644 (file)
--- a/psycopg3/_pq_ctypes.py
+++ b/psycopg3/_pq_ctypes.py
@@ -7,7 +7,7 @@ libpq access using ctypes
 import ctypes
 import ctypes.util
 from ctypes import Structure, POINTER
-from ctypes import c_char_p, c_int
+from ctypes import c_char_p, c_int, c_void_p
 
 pq = ctypes.pydll.LoadLibrary(ctypes.util.find_library("pq"))
 
@@ -70,6 +70,10 @@ PQconninfo = pq.PQconninfo
 PQconninfo.argtypes = [PGconn_ptr]
 PQconninfo.restype = PQconninfoOption_ptr
 
+PQconninfoParse = pq.PQconninfoParse
+PQconninfoParse.argtypes = [c_char_p, POINTER(c_char_p)]
+PQconninfoParse.restype = PQconninfoOption_ptr
+
 PQfinish = pq.PQfinish
 PQfinish.argtypes = [PGconn_ptr]
 PQfinish.restype = None
@@ -104,3 +108,10 @@ PQerrorMessage.restype = c_char_p
 PQsocket = pq.PQsocket
 PQsocket.argtypes = [PGconn_ptr]
 PQsocket.restype = c_int
+
+
+# 33.11. Miscellaneous Functions
+
+PQfreemem = pq.PQfreemem
+PQfreemem.argtypes = [c_void_p]
+PQfreemem.restype = None

diff --git a/psycopg3/pq.py b/psycopg3/pq.py
index ecc6a9f7bc1e3cfffe8114a13e6b426e0f2f4341..f97a3a5629af1c04cd9c066b6101a9f1df2cb356 100644 (file)
--- a/psycopg3/pq.py
+++ b/psycopg3/pq.py
@@ -14,5 +14,6 @@ from .pq_enums import ConnStatus
 from . import pq_ctypes as pq_module
 
 PGconn = pq_module.PGconn
+PQerror = pq_module.PQerror
 
-__all__ = ("ConnStatus", "PGconn")
+__all__ = ("ConnStatus", "PGconn", "PQerror")

diff --git a/psycopg3/pq_ctypes.py b/psycopg3/pq_ctypes.py
index 5e784e00be8dd4a26b4c34aebedef129c563a420..b8937d912988c88a41b5cf4038e7613c517ea9c7 100644 (file)
--- a/psycopg3/pq_ctypes.py
+++ b/psycopg3/pq_ctypes.py
@@ -9,6 +9,7 @@ implementation.
 # Copyright (C) 2020 The Psycopg Team
 
 from collections import namedtuple
+from ctypes import c_char_p, pointer
 
 from .pq_enums import ConnStatus, PostgresPollingStatus, PGPing
 from . import _pq_ctypes as impl
@@ -44,7 +45,6 @@ class PGconn:
     def connect_start(cls, conninfo):
         if isinstance(conninfo, str):
             conninfo = conninfo.encode("utf8")
-
         if not isinstance(conninfo, bytes):
             raise TypeError("bytes expected, got %r instead" % conninfo)
 
@@ -75,6 +75,28 @@ class PGconn:
         finally:
             impl.PQconninfoFree(opts)
 
+    @classmethod
+    def parse_conninfo(cls, conninfo):
+        if isinstance(conninfo, str):
+            conninfo = conninfo.encode("utf8")
+        if not isinstance(conninfo, bytes):
+            raise TypeError("bytes expected, got %r instead" % conninfo)
+
+        errmsg = c_char_p()
+        rv = impl.PQconninfoParse(conninfo, pointer(errmsg))
+        if not rv:
+            if not errmsg:
+                raise MemoryError("couldn't allocate on conninfo parse")
+            else:
+                exc = PQerror(errmsg.value.decode("utf8", "replace"))
+                impl.PQfreemem(errmsg)
+                raise exc
+
+        try:
+            return _conninfoopts_from_array(rv)
+        finally:
+            impl.PQconninfoFree(rv)
+
     def reset(self):
         impl.PQreset(self.pgconn_ptr)
 
@@ -96,6 +118,8 @@ class PGconn:
     def ping(self, conninfo):
         if isinstance(conninfo, str):
             conninfo = conninfo.encode("utf8")
+        if not isinstance(conninfo, bytes):
+            raise TypeError("bytes expected, got %r instead" % conninfo)
 
         rv = impl.PQping(conninfo)
         return PGPing(rv)

diff --git a/tests/test_pq.py b/tests/test_pq.py
index c524ea62f15ace29036d01792b75170c755423c0..79f44106a691d01f81868259504cd7683200ba6e 100644 (file)
--- a/tests/test_pq.py
+++ b/tests/test_pq.py
@@ -94,6 +94,24 @@ def test_info(pq, dsn):
     assert dbname.dispsize == 20
 
 
+def test_conninfo_parse(pq):
+    info = pq.PGconn.parse_conninfo(
+        "postgresql://host1:123,host2:456/somedb"
+        "?target_session_attrs=any&application_name=myapp"
+    )
+    info = {i.keyword: i.val for i in info if i.val is not None}
+    assert info["host"] == "host1,host2"
+    assert info["port"] == "123,456"
+    assert info["dbname"] == "somedb"
+    assert info["application_name"] == "myapp"
+
+
+def test_conninfo_parse_bad(pq):
+    with pytest.raises(pq.PQerror) as e:
+        pq.PGconn.parse_conninfo("bad_conninfo=")
+        assert "bad_conninfo" in str(e.value)
+
+
 def test_reset(pq, dsn):
     conn = pq.PGconn.connect(dsn)
     assert conn.status == ConnStatus.CONNECTION_OK