Distinguish jsonb[] array columns from scalar jsonb in ComplexHelper

pgdevkit/db/complex_types.py

@@ -9,7 +9,18 @@
 from psycopg.types.enum import EnumInfo, register_enum
 from psycopg.types.json import Jsonb
 
-ComplexTypeInfo = CompositeInfo | EnumInfo | type[Jsonb] | None
+
+class JsonbArray:
+    """Sentinel for a `jsonb[]` column (a Postgres ARRAY of jsonb) — distinct
+    from `Jsonb` (a scalar jsonb column) so `recursive_convert` knows whether
+    an incoming Python list *is* the array (each element needs its own
+    `Jsonb(...)` wrapper) or is JSON array *content* for one scalar jsonb
+    value (the whole list gets wrapped once). Both cases arrive from
+    `information_schema.columns` looking identical (a plain Python list) —
+    only the column's own array-ness disambiguates them."""
+
+
+ComplexTypeInfo = CompositeInfo | EnumInfo | type[Jsonb] | type[JsonbArray] | None
 
 
 class ComplexHelper:
@@ -67,7 +78,7 @@ async def _load_complex_type_from_colinfos(self, res: dict[str, Any] | None) ->
         if res["data_type"].lower() == "jsonb":
             return Jsonb
         if res["data_type"].upper() == "ARRAY" and res["udt_name"] == "_jsonb":
-            return Jsonb
+            return JsonbArray
         if (
             not res["is_enum"]
             and not res["is_user_defined"]
@@ -181,9 +192,16 @@ async def recursive_convert(
             return None
         if self.system_complex_type_dict is None:
             await self.load_complex_type_dict()
+        if info == JsonbArray:
+            # The value here *is* the jsonb[] array -- each element is its
+            # own scalar jsonb value, unlike the plain-Jsonb case below.
+            assert isinstance(value, list), f"Expected a list for a jsonb[] column, got {type(value)}"
+            return [Jsonb(item) for item in value]
         if info == Jsonb:
-            # A JSONB column's value is wrapped whole, even if it's a list —
-            # only an array-of-composite/enum column recurses per element.
+            # A scalar JSONB column's value is wrapped whole, even if it's a
+            # list — that list is JSON array *content* for the one jsonb
+            # value, not multiple array elements (see JsonbArray above for
+            # the jsonb[] column case).
             return Jsonb(value)
         if isinstance(value, list):
             return [await self.recursive_convert(item, info, con) for item in value]

pyproject.toml

@@ -11,7 +11,7 @@ packages = ["pgdevkit"]
 
 [project]
 name = "pgdevkit"
-version = "0.2.3"
+version = "0.2.4"
 description = "A helper for developing with Postgres"
 readme = "README.md"
 requires-python = ">=3.14"

tests/db/test_complex_types.py

@@ -80,6 +80,47 @@ async def test_select_list_includes_generated_columns(complex_types_test_db):
     assert '"total"' in rendered
 
 
+@requires_podman
+async def test_jsonb_array_column_wraps_each_element_not_the_whole_list(complex_types_test_db):
+    # Regression: a jsonb[] column (a Postgres ARRAY of jsonb) and a plain
+    # jsonb column both surface identically from information_schema (a bare
+    # Python list, if that's the value) -- previously both were detected as
+    # the same `Jsonb` sentinel, and recursive_convert wrapped the whole
+    # incoming list as one Jsonb(...) for either case. That's correct for a
+    # plain jsonb column storing a JSON array *value*, but wrong for a
+    # jsonb[] column, where the list *is* the array and each element needs
+    # its own Jsonb(...) wrapper -- psycopg would otherwise try to bind a
+    # single jsonb value to an array column and Postgres would raise
+    # DatatypeMismatch ("column ... is of type jsonb[] but expression is of
+    # type jsonb").
+    async with await psycopg.AsyncConnection.connect(_db_dsn(), autocommit=True) as con:
+        await con.execute("""
+            CREATE TABLE gadget (
+                id serial PRIMARY KEY,
+                tags jsonb[],
+                metadata jsonb
+            )
+        """)
+        helper = ComplexHelper(con)
+        types = await helper.load_all_complex_types(("public", "gadget"))
+
+        tags_value = [{"name": "a"}, {"name": "b"}]
+        converted_tags = await helper.recursive_convert(tags_value, types["tags"], con)
+
+        metadata_value = [1, 2, 3]  # JSON array *content* for one scalar jsonb value
+        converted_metadata = await helper.recursive_convert(metadata_value, types["metadata"], con)
+
+        await con.execute(
+            "INSERT INTO gadget (id, tags, metadata) VALUES (1, %s, %s)",
+            (converted_tags, converted_metadata),
+        )
+        async with con.cursor() as cur:
+            await cur.execute("SELECT tags, metadata FROM gadget WHERE id = 1")
+            tags, metadata = await cur.fetchone()
+    assert tags == tags_value
+    assert metadata == metadata_value
+
+
 def test_complex_types_cache_is_per_instance_not_shared():
     # Regression: complex_types used to be a mutable class attribute, so a
     # CompositeInfo/EnumInfo (which carries OIDs from one specific