fix: correctly map execute_many parameters for all drivers (#264)

Normalize mapping batches before parameter alignment so `execute_many` with list-of-dicts works with positional execution styles.

  - The Problem: DuckDB (and other positional adapters) raised identifier/
    count mismatches when execute_many received a list of dicts for named
    placeholders.
  - The Solution: Core parameter processor now converts mapping payloads
    to ordered positional values prior to validation; converter preserves
    unnamed placeholder ordinals; added regression tests.

Author: cofin

sqlspec/core/parameters/_converter.py

@@ -79,13 +79,13 @@ def convert_placeholder_style(
         current_styles = {p.style for p in param_info}
         if len(current_styles) == 1 and target_style in current_styles:
             converted_parameters = self._convert_parameter_format(
-                parameters, param_info, target_style, parameters, preserve_parameter_format=True
+                parameters, param_info, target_style, parameters, preserve_parameter_format=True, is_many=is_many
             )
             return sql, converted_parameters
 
         converted_sql = self._convert_placeholders_to_style(sql, param_info, target_style)
         converted_parameters = self._convert_parameter_format(
-            parameters, param_info, target_style, parameters, preserve_parameter_format=True
+            parameters, param_info, target_style, parameters, preserve_parameter_format=True, is_many=is_many
         )
         return converted_sql, converted_parameters
 
@@ -230,10 +230,30 @@ def _convert_parameter_format(
         target_style: "ParameterStyle",
         original_parameters: Any = None,
         preserve_parameter_format: bool = False,
+        is_many: bool = False,
     ) -> Any:
         if not parameters or not param_info:
             return parameters
 
+        if (
+            is_many
+            and isinstance(parameters, Sequence)
+            and not isinstance(parameters, (str, bytes, bytearray))
+            and parameters
+            and isinstance(parameters[0], Mapping)
+        ):
+            normalized_sets = [
+                self._convert_parameter_format(
+                    param_set, param_info, target_style, param_set, preserve_parameter_format, is_many=False
+                )
+                if isinstance(param_set, Mapping)
+                else param_set
+                for param_set in parameters
+            ]
+            if preserve_parameter_format and isinstance(parameters, tuple):
+                return tuple(normalized_sets)
+            return normalized_sets
+
         is_named_style = target_style in {
             ParameterStyle.NAMED_COLON,
             ParameterStyle.NAMED_AT,
@@ -261,15 +281,15 @@ def _convert_parameter_format(
             if has_mixed_styles:
                 param_keys = list(parameters.keys())
                 for param in param_info:
-                    param_key = param.placeholder_text
+                    param_key = param.placeholder_text if param.name else f"{param.placeholder_text}_{param.ordinal}"
                     if param_key not in unique_params:
                         value, found = self._extract_param_value_mixed_styles(param, parameters, param_keys)
                         if found:
                             unique_params[param_key] = value
                             param_order.append(param_key)
             else:
                 for param in param_info:
-                    param_key = param.placeholder_text
+                    param_key = param.placeholder_text if param.name else f"{param.placeholder_text}_{param.ordinal}"
                     if param_key not in unique_params:
                         value, found = self._extract_param_value_single_style(param, parameters)
                         if found:

sqlspec/core/parameters/_processor.py

@@ -6,6 +6,7 @@
 
 from mypy_extensions import mypyc_attr
 
+from sqlspec.core.parameters._alignment import looks_like_execute_many
 from sqlspec.core.parameters._converter import ParameterConverter
 from sqlspec.core.parameters._types import (
     ParameterInfo,
@@ -169,6 +170,35 @@ def process(
 
         needs_static_embedding = config.needs_static_script_compilation and param_info and parameters and not is_many
 
+        def _requires_mapping_normalization(payload: Any) -> bool:
+            if not payload or not param_info:
+                return False
+
+            has_named_placeholders = any(
+                param.style
+                in {
+                    ParameterStyle.NAMED_COLON,
+                    ParameterStyle.NAMED_AT,
+                    ParameterStyle.NAMED_DOLLAR,
+                    ParameterStyle.NAMED_PYFORMAT,
+                }
+                for param in param_info
+            )
+            if has_named_placeholders:
+                return False
+
+            looks_many = is_many or looks_like_execute_many(payload)
+            if not looks_many:
+                return False
+
+            if isinstance(payload, Mapping):
+                return True
+
+            if isinstance(payload, Sequence) and not isinstance(payload, (str, bytes, bytearray)):
+                return any(isinstance(item, Mapping) for item in payload)
+
+            return False
+
         if needs_static_embedding:
             return self._handle_static_embedding(sql, parameters, config, is_many, cache_key)
 
@@ -177,6 +207,7 @@ def process(
             and not needs_execution_conversion
             and not config.type_coercion_map
             and not config.output_transformer
+            and not _requires_mapping_normalization(parameters)
         ):
             result = ParameterProcessingResult(sql, parameters, ParameterProfile(param_info))
             if self._cache_size < self.DEFAULT_CACHE_SIZE:
@@ -186,6 +217,12 @@ def process(
 
         processed_sql, processed_parameters = sql, parameters
 
+        if _requires_mapping_normalization(processed_parameters):
+            target_style = self._determine_target_execution_style(original_styles, config)
+            processed_sql, processed_parameters = self._converter.convert_placeholder_style(
+                processed_sql, processed_parameters, target_style, is_many
+            )
+
         if processed_parameters:
             processed_parameters = self._apply_type_wrapping(processed_parameters)
 

tests/unit/test_core/test_parameters.py

@@ -1026,6 +1026,41 @@ def test_process_full_pipeline(processor: ParameterProcessor) -> None:
         assert len(final_params) > 0
 
 
+def test_process_execute_many_mapping_payload(
+    processor: "ParameterProcessor", basic_config: "ParameterStyleConfig"
+) -> None:
+    """Ensure execute_many normalizes mapping payloads for positional placeholders."""
+
+    sql = "INSERT INTO metrics (a, b) VALUES (?, ?)"
+    parameters = [{"a": 1, "b": "x"}, {"a": 2, "b": "y"}]
+
+    final_sql, final_params = processor.process(sql, parameters, basic_config, is_many=True)
+
+    assert final_sql == sql
+    assert isinstance(final_params, list)
+    assert all(isinstance(param_set, (list, tuple)) for param_set in final_params)
+    assert [tuple(param_set) for param_set in final_params] == [(1, "x"), (2, "y")]
+
+
+def test_process_execute_many_named_to_positional(processor: "ParameterProcessor") -> None:
+    """Execute_many with named placeholders should convert mapping batches to positional values."""
+
+    config = ParameterStyleConfig(
+        default_parameter_style=ParameterStyle.NAMED_DOLLAR,
+        supported_parameter_styles={ParameterStyle.NAMED_DOLLAR, ParameterStyle.QMARK},
+        default_execution_parameter_style=ParameterStyle.QMARK,
+    )
+
+    sql = "INSERT INTO metrics (a, b) VALUES ($a, $b)"
+    parameters = [{"a": 10, "b": 20}, {"b": 40, "a": 30}]
+
+    final_sql, final_params = processor.process(sql, parameters, config, "duckdb", is_many=True)
+
+    assert final_sql.count("?") == 2
+    assert isinstance(final_params, list)
+    assert [tuple(param_set) for param_set in final_params] == [(10, 20), (30, 40)]
+
+
 def test_list_parameter_preservation(converter: ParameterConverter) -> None:
     """Test that list parameters are properly handled."""
     sql = "INSERT INTO users (id, name, active) VALUES (?, ?, ?)"