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

src/json2sql/converter.py

@@ -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)

tests/test_edge_cases.py

@@ -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

tests/test_type_inference.py

@@ -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