Added composite.fetch_info_async

diff --git a/psycopg3/types/composite.py b/psycopg3/types/composite.py
index d7e5161ab37cfac19b50b835ebdb503b621e9f91..8ff8d33f76afda1bf606b46fa64a3d1484a91761 100644 (file)
--- a/psycopg3/types/composite.py
+++ b/psycopg3/types/composite.py
@@ -13,7 +13,7 @@ from ..adapt import Format, TypeCaster, Transformer, AdaptContext
 from .oids import builtins, TypeInfo
 
 if TYPE_CHECKING:
-    from ..connection import Connection
+    from ..connection import Connection, AsyncConnection
 
 
 TEXT_OID = builtins["text"].oid
@@ -48,10 +48,16 @@ def fetch_info(conn: "Connection", name: str) -> Optional[CompositeTypeInfo]:
     cur = conn.cursor(binary=True)
     cur.execute(_type_info_query, (name,))
     rec = cur.fetchone()
-    if rec is not None:
-        return CompositeTypeInfo(*rec)
-    else:
-        return None
+    return CompositeTypeInfo(*rec) if rec is not None else None
+
+
+async def fetch_info_async(
+    conn: "AsyncConnection", name: str
+) -> Optional[CompositeTypeInfo]:
+    cur = conn.cursor(binary=True)
+    await cur.execute(_type_info_query, (name,))
+    rec = await cur.fetchone()
+    return CompositeTypeInfo(*rec) if rec is not None else None
 
 
 def register(

diff --git a/tests/types/test_composite.py b/tests/types/test_composite.py
index 32e3248e4937705dfc2b69fdb1cda9dd846a631d..acaef811cfb7f8742e783b74e875b39a491e9508 100644 (file)
--- a/tests/types/test_composite.py
+++ b/tests/types/test_composite.py
@@ -100,6 +100,21 @@ def test_fetch_info(conn, testcomp):
         assert info.fields[i].type_oid == builtins[t].oid
 
 
+def test_fetch_info_async(aconn, loop, testcomp):
+    info = loop.run_until_complete(
+        composite.fetch_info_async(aconn, "testcomp")
+    )
+    assert info.name == "testcomp"
+    assert info.oid > 0
+    assert info.oid != info.array_oid > 0
+    assert len(info.fields) == 3
+    for i, (name, t) in enumerate(
+        [("foo", "text"), ("bar", "int8"), ("baz", "float8")]
+    ):
+        assert info.fields[i].name == name
+        assert info.fields[i].type_oid == builtins[t].oid
+
+
 @pytest.mark.parametrize("fmt_out", [Format.TEXT, Format.BINARY])
 def test_cast_composite(conn, testcomp, fmt_out):
     cur = conn.cursor(binary=fmt_out == Format.BINARY)