cowork-bot: automated improvements (cowork/improve-json2sql-3) (#39)

* cowork-bot: dedupe nested-array child tables when multiple parent rows carry arrays

Grouping children per key so convert()/generate_schema() emit exactly one
CREATE TABLE per child table, with every child row linked to its own parent
FK (previously duplicate CREATE TABLEs and dropped rows).

* cowork-bot: fix FK column detection in flatten mode

When flattening nested arrays, the FK column in the child table must match
the parent table's primary key column name and type. Previously the code
preferred 'name' over explicit ID fields like 'user_id' or 'users_id',
causing a type mismatch (TEXT FK vs INTEGER PK).

New priority order for parent reference key:
1. 'id' (generic primary key)
2. '{parent_table}_id' (table-specific, e.g., 'users_id')
3. Any key ending in '_id' found in parent objects (e.g., 'user_id')
4. 'name' (fallback only when no ID-like field exists)

Added 12 regression tests covering all three dialects (Postgres, MySQL, SQLite).

---------

Co-authored-by: Jaixii <algorithmictradingsolutions@gmail.com>

Author: github-actions[bot]

src/json2sql.egg-info/PKG-INFO

@@ -93,15 +93,22 @@ def _convert_objects(self, objects: list[dict], table_name: str) -> str:
         # When flattening, compute the full column set first so rows align
         if self.flatten:
             columns, flat_map = self._infer_columns_flattened(objects, table_name)
-            # Process nested arrays into child tables
+            # Process nested arrays into child tables, grouped by key so that
+            # each nested array produces exactly ONE child table whose INSERT
+            # covers every parent row's children.
+            nested_groups: dict[str, tuple[list[dict], list[dict]]] = {}
             for obj in objects:
                 for key, value in obj.items():
                     if (
                         isinstance(value, list)
                         and value
                         and all(isinstance(v, dict) for v in value)
                     ):
-                        self._flatten_nested(table_name, key, value, obj)
+                        children, parents = nested_groups.setdefault(key, ([], []))
+                        children.extend(value)
+                        parents.extend([obj] * len(value))
+            for key, (children, parents) in nested_groups.items():
+                self._flatten_nested(table_name, key, children, parents)
         else:
             columns = self._infer_columns(objects)
             flat_map = {}
@@ -240,27 +247,50 @@ def _flatten_nested(
         parent_table: str,
         key: str,
         nested_objects: list[dict],
-        parent_obj: dict,
+        parent_objs: list[dict],
     ) -> None:
-        """Flatten a nested array of objects into a separate table."""
+        """Flatten nested arrays of objects into a single child table.
+
+        ``nested_objects`` and ``parent_objs`` are aligned lists: each child
+        row links back to its own parent via the foreign key. Grouping all
+        parents' children into one table avoids emitting duplicate
+        ``CREATE TABLE`` statements when multiple rows carry nested arrays.
+        """
         child_table = f"{parent_table}_{key}"
         columns = self._infer_columns(nested_objects)
-        # Add parent reference — only if no existing column has the FK name
+        # Add parent reference — only if no existing column has the FK name.
+        # Prefer explicit ID fields over generic "name" to ensure the FK column
+        # type matches the parent table's primary key type.
         parent_ref = None
-        for pk in ("id", "name", parent_table + "_id"):
-            if pk in parent_obj:
+        # Priority order for parent reference key:
+        # 1. "id" (generic primary key)
+        # 2. "{parent_table}_id" (table-specific, e.g., "users_id")
+        # 3. Any key ending in "_id" found in parent objects (e.g., "user_id")
+        # 4. "name" (fallback only when no ID-like field exists)
+        candidate_keys = ["id", f"{parent_table}_id"]
+        # Add any *_id keys found in parent objects (excluding already listed)
+        seen = set(candidate_keys)
+        for obj in parent_objs:
+            for k in obj:
+                if k.endswith("_id") and k not in seen:
+                    candidate_keys.append(k)
+                    seen.add(k)
+        candidate_keys.append("name")
+        for pk in candidate_keys:
+            if any(pk in parent_obj for parent_obj in parent_objs):
                 parent_ref = pk
                 break
         fk_col = f"{parent_table}_{parent_ref}" if parent_ref else None
         fk_already_exists = fk_col and fk_col in columns
         if fk_col and not fk_already_exists:
+            fk_parent = next(p for p in parent_objs if parent_ref in p)
             columns = {
-                fk_col: sql_type_for(parent_obj[parent_ref], self.dialect),
+                fk_col: sql_type_for(fk_parent[parent_ref], self.dialect),
                 **columns,
             }
 
         rows: list[list[str]] = []
-        for nested in nested_objects:
+        for nested, parent_obj in zip(nested_objects, parent_objs, strict=True):
             row: list[str] = []
             for col_name in columns:
                 if col_name == fk_col and not fk_already_exists:
@@ -277,11 +307,16 @@ def _process_flatten(self, objects: list, table_name: str) -> None:
             return
         if not objects or not isinstance(objects[0], dict):
             return
+        nested_groups: dict[str, tuple[list[dict], list[dict]]] = {}
         for obj in objects:
             for key, value in obj.items():
                 if (
                     isinstance(value, list)
                     and value
                     and all(isinstance(v, dict) for v in value)
                 ):
-                    self._flatten_nested(table_name, key, value, obj)
+                    children, parents = nested_groups.setdefault(key, ([], []))
+                    children.extend(value)
+                    parents.extend([obj] * len(value))
+        for key, (children, parents) in nested_groups.items():
+            self._flatten_nested(table_name, key, children, parents)

src/json2sql.egg-info/SOURCES.txt

@@ -108,3 +108,41 @@ def test_convert_objects_list_vs_dict_root(self):
         result = converter.convert(json.dumps([{"name": "test"}]))
         assert "INSERT INTO" in result
         assert "'test'" in result
+
+
+def test_flatten_multiple_parent_rows_single_child_table():
+    """Multiple parent rows with nested arrays yield ONE child table with all rows."""
+    import json as _json
+
+    from json2sql.converter import JSONToSQLConverter
+
+    data = [
+        {"id": 1, "name": "a", "tags": [{"label": "x", "score": 1}]},
+        {
+            "id": 2,
+            "name": "b",
+            "tags": [{"label": "y", "score": 2}, {"label": "z", "score": 3}],
+        },
+    ]
+    text = _json.dumps(data)
+    out = JSONToSQLConverter(flatten=True).convert(text, "users")
+    assert out.count('CREATE TABLE "users_tags"') == 1
+    assert "'z', 3" in out and "'y', 2" in out and "'x', 1" in out
+
+    schema = JSONToSQLConverter(flatten=True).generate_schema(text, "users")
+    assert schema.count('CREATE TABLE "users_tags"') == 1
+
+
+def test_flatten_child_rows_keep_own_parent_fk():
+    """Each child row links to its own parent via the FK column."""
+    import json as _json
+
+    from json2sql.converter import JSONToSQLConverter
+
+    data = [
+        {"id": 10, "items": [{"sku": "a1"}]},
+        {"id": 20, "items": [{"sku": "b1"}]},
+    ]
+    out = JSONToSQLConverter(flatten=True).convert(_json.dumps(data), "orders")
+    assert "(10, 'a1')" in out
+    assert "(20, 'b1')" in out

src/json2sql.egg-info/dependency_links.txt

@@ -131,3 +131,72 @@ def test_convert_never_emits_empty_column_list():
     )
     assert "();" not in out
     assert "INSERT INTO" in out
+
+
+class TestFlattenFKDetection:
+    """Tests for correct FK column detection in flatten mode.
+
+    The FK column in a child table must match the parent table's primary key
+    column name and type. Previously the code preferred "name" over explicit
+    ID fields like "user_id" or "users_id", causing a type mismatch.
+    """
+
+    @pytest.mark.parametrize("dialect", [Dialect.POSTGRES, Dialect.MYSQL, Dialect.SQLITE])
+    def test_flatten_prefers_id_over_name(self, dialect):
+        """When parent has both 'id' and 'name', 'id' should be used for FK."""
+        conv = JSONToSQLConverter(dialect=dialect, flatten=True)
+        data = json.dumps([
+            {"id": 1, "name": "Alice", "tags": [{"label": "x"}]},
+            {"id": 2, "name": "Bob", "tags": [{"label": "y"}]},
+        ])
+        out = conv.convert(data, table_name="users")
+        # FK column should be users_id (from parent's id), not users_name
+        assert '"users_id"' in out or '`users_id`' in out
+        assert '"users_name"' not in out and '`users_name`' not in out
+        # Parent table should have id column
+        assert '"id"' in out or '`id`' in out
+
+    @pytest.mark.parametrize("dialect", [Dialect.POSTGRES, Dialect.MYSQL, Dialect.SQLITE])
+    def test_flatten_prefers_table_specific_id(self, dialect):
+        """When parent has '{table}_id' (e.g., users_id), it should be used."""
+        conv = JSONToSQLConverter(dialect=dialect, flatten=True)
+        data = json.dumps([
+            {"users_id": 10, "name": "Alice", "tags": [{"label": "x"}]},
+            {"users_id": 20, "name": "Bob", "tags": [{"label": "y"}]},
+        ])
+        out = conv.convert(data, table_name="users")
+        # FK column should be users_users_id (from parent's users_id)
+        assert '"users_users_id"' in out or '`users_users_id`' in out
+        assert '"users_name"' not in out and '`users_name`' not in out
+
+    @pytest.mark.parametrize("dialect", [Dialect.POSTGRES, Dialect.MYSQL, Dialect.SQLITE])
+    def test_flatten_prefers_any_id_suffix(self, dialect):
+        """When parent has a singular '*_id' (e.g., user_id), it should be used over 'name'."""
+        conv = JSONToSQLConverter(dialect=dialect, flatten=True)
+        data = json.dumps([
+            {"user_id": 100, "name": "Alice", "tags": [{"label": "x"}]},
+            {"user_id": 200, "name": "Bob", "tags": [{"label": "y"}]},
+        ])
+        out = conv.convert(data, table_name="users")
+        # FK column should be users_user_id (from parent's user_id)
+        assert '"users_user_id"' in out or '`users_user_id`' in out
+        assert '"users_name"' not in out and '`users_name`' not in out
+        # FK type should be numeric (matching parent's user_id type)
+        if dialect == Dialect.MYSQL:
+            assert "INT" in out
+        else:
+            assert "INTEGER" in out
+
+    @pytest.mark.parametrize("dialect", [Dialect.POSTGRES, Dialect.MYSQL, Dialect.SQLITE])
+    def test_flatten_fallback_to_name_when_no_id(self, dialect):
+        """When parent has no ID-like field, 'name' is used as fallback."""
+        conv = JSONToSQLConverter(dialect=dialect, flatten=True)
+        data = json.dumps([
+            {"name": "Alice", "tags": [{"label": "x"}]},
+            {"name": "Bob", "tags": [{"label": "y"}]},
+        ])
+        out = conv.convert(data, table_name="users")
+        # FK column should be users_name (fallback)
+        assert '"users_name"' in out or '`users_name`' in out
+        # Parent table should have name column
+        assert '"name"' in out or '`name`' in out