"""Parameter processing pipeline orchestrator."""
from collections import OrderedDict
from collections.abc import Callable, Mapping, Sequence
from typing import Any, Final, cast
from mypy_extensions import mypyc_attr
from sqlspec.core.parameters._converter import (
ParameterConverter,
_named_parameter_names,
_parameter_lookup_key,
ambiguous_index_message,
ambiguous_index_placeholder,
)
from sqlspec.core.parameters._types import (
_EXPANDING_POSITIONAL_STYLES,
_NAMED_STYLE_VALUES,
_NAMED_STYLES,
_POSITIONAL_STYLE_VALUES,
ConvertedParameters,
ParameterInfo,
ParameterPayload,
ParameterProcessingResult,
ParameterProfile,
ParameterStyle,
ParameterStyleConfig,
TypedParameter,
wrap_with_type,
)
from sqlspec.core.parameters._validator import ParameterValidator, unescape_qmark_operator
from sqlspec.exceptions import SQLSpecError
from sqlspec.utils.dispatch import TypeDispatcher
_OCCURRENCE_KEYED_STYLE_VALUES: "frozenset[str]" = frozenset({
ParameterStyle.QMARK.value,
ParameterStyle.POSITIONAL_PYFORMAT.value,
})
__all__ = (
"ParameterProcessor",
"apply_type_coercion",
"structural_fingerprint",
"type_coercion_dispatcher",
"type_coercion_fallbacks",
"value_fingerprint",
)
TypeCoercionFallback = tuple[type, Callable[[Any], Any]]
_EXECUTE_MANY_SAMPLE_THRESHOLD: Final[int] = 10
_EXECUTE_MANY_SAMPLE_SIZE: Final[int] = 3
_TYPE_COERCION_DISPATCHERS: Final[dict[tuple[TypeCoercionFallback, ...], TypeDispatcher[Callable[[Any], Any]]]] = {}
def structural_fingerprint(parameters: "ParameterPayload", is_many: bool = False) -> Any:
"""Return a structural fingerprint for caching parameter payloads.
Returns a hashable tuple representing the structure (keys, types, count).
Avoids string formatting for performance.
Note: Uses Python 3.7+ dict insertion order instead of sorted() for determinism.
This means fingerprints depend on the order keys were inserted, which is typically
consistent within a single codebase.
"""
if parameters is None:
return None
# Fast type dispatch: check concrete types first
param_type = type(parameters)
# Handle dict (most common Mapping type) - fast path
if param_type is dict:
dict_params = cast("dict[str, Any]", parameters)
if not dict_params:
return ("dict",)
# Use dict insertion order (Python 3.7+ guaranteed) instead of sorted()
# This is O(n) vs O(n log n) and produces consistent fingerprints for
# parameters constructed in the same order (typical usage pattern)
keys = tuple(dict_params.keys())
type_sig = tuple(type(v) for v in dict_params.values())
return ("dict", keys, type_sig)
# Handle list and tuple (most common Sequence types) - fast path
if param_type is list or param_type is tuple:
seq_params = cast("Sequence[Any]", parameters)
if not seq_params:
return ("seq",)
# Optimization: Fast path for single-item sequence (extremely common)
if len(seq_params) == 1:
return ("seq", (type(seq_params[0]),))
if is_many:
return _fingerprint_execute_many(seq_params)
# Single execution with sequence parameters
type_sig = tuple(type(v) for v in seq_params)
return ("seq", type_sig)
# Fallback to ABC checks for custom types (Mapping, Sequence subclasses)
if isinstance(parameters, Mapping):
if not parameters:
return ("dict",)
keys = tuple(parameters.keys())
type_sig = tuple(type(v) for v in parameters.values())
return ("dict", keys, type_sig)
if isinstance(parameters, Sequence) and not isinstance(parameters, (str, bytes, bytearray)):
if not parameters:
return ("seq",)
if len(parameters) == 1:
return ("seq", (type(parameters[0]),))
if is_many:
return _fingerprint_execute_many(parameters)
type_sig = tuple(type(v) for v in parameters)
return ("seq", type_sig)
# Scalar parameter
return ("scalar", param_type)
def value_fingerprint(parameters: "ParameterPayload") -> Any:
"""Return a value-based fingerprint for parameter payloads.
Unlike structural_fingerprint, this includes actual parameter VALUES in the hash.
Used for static script compilation where SQL has values embedded directly.
Args:
parameters: Original parameter payload supplied by the caller.
Returns:
Hashable representation including parameter values.
"""
if parameters is None:
return None
# Use repr for value-based hashing - includes both structure and values
# Return as tuple to match structural_fingerprint return type (hashable)
return ("values", repr(parameters))
[docs]
@mypyc_attr(allow_interpreted_subclasses=False)
class ParameterProcessor:
"""Parameter processing engine coordinating conversion phases."""
__slots__ = ("_cache", "_cache_hits", "_cache_max_size", "_cache_misses", "_converter", "_validator")
DEFAULT_CACHE_SIZE = 1000
[docs]
def __init__(
self,
*,
converter: "ParameterConverter | None" = None,
validator: "ParameterValidator | None" = None,
cache_max_size: int | None = None,
validator_cache_max_size: int | None = None,
) -> None:
self._cache: OrderedDict[Any, ParameterProcessingResult] = OrderedDict()
if cache_max_size is None:
cache_max_size = self.DEFAULT_CACHE_SIZE
self._cache_max_size = max(cache_max_size, 0)
self._cache_hits = 0
self._cache_misses = 0
if converter is None:
if validator is None:
validator_cache = validator_cache_max_size
if validator_cache is None:
validator_cache = self._cache_max_size
validator = ParameterValidator(cache_max_size=validator_cache)
self._validator = validator
self._converter = ParameterConverter(self._validator)
else:
self._converter = converter
if validator is None:
self._validator = converter.validator
else:
self._validator = validator
self._converter.validator = validator
if validator_cache_max_size is not None and isinstance(self._validator, ParameterValidator):
self._validator.set_cache_max_size(validator_cache_max_size)
[docs]
def clear_cache(self) -> None:
"""Clear cached processing results and reset stats."""
self._cache.clear()
self._cache_hits = 0
self._cache_misses = 0
if isinstance(self._validator, ParameterValidator):
self._validator.clear_cache()
[docs]
def cache_stats(self) -> "dict[str, int]":
"""Return cache statistics for parameter processing."""
stats = {
"hits": self._cache_hits,
"misses": self._cache_misses,
"size": len(self._cache),
"max_size": self._cache_max_size,
}
if isinstance(self._validator, ParameterValidator):
validator_stats = self._validator.cache_stats()
stats["validator_hits"] = validator_stats["hits"]
stats["validator_misses"] = validator_stats["misses"]
stats["validator_size"] = validator_stats["size"]
stats["validator_max_size"] = validator_stats["max_size"]
else:
stats["validator_hits"] = 0
stats["validator_misses"] = 0
stats["validator_size"] = 0
stats["validator_max_size"] = 0
return stats
def process(
self,
sql: str,
parameters: "ParameterPayload",
config: "ParameterStyleConfig",
dialect: str | None = None,
is_many: bool = False,
wrap_types: bool = True,
param_fingerprint: Any | None = None,
) -> "ParameterProcessingResult":
return self._process_internal(
sql,
parameters,
config,
dialect=dialect,
is_many=is_many,
wrap_types=wrap_types,
normalize_for_parsing=True,
param_fingerprint=param_fingerprint,
)
[docs]
def process_for_execution(
self,
sql: str,
parameters: "ParameterPayload",
config: "ParameterStyleConfig",
dialect: str | None = None,
is_many: bool = False,
wrap_types: bool = True,
parsed_expression: Any = None,
param_fingerprint: Any | None = None,
) -> "ParameterProcessingResult":
"""Process parameters for execution without parse normalization.
Args:
sql: SQL string to process.
parameters: Parameter payload.
config: Parameter style configuration.
dialect: Optional SQL dialect.
is_many: Whether this is execute_many.
wrap_types: Whether to wrap parameters with type metadata.
parsed_expression: Pre-parsed SQLGlot expression to preserve through pipeline.
param_fingerprint: Pre-computed parameter fingerprint for cache key.
Returns:
ParameterProcessingResult with execution SQL and parameters.
"""
return self._process_internal(
sql,
parameters,
config,
dialect=dialect,
is_many=is_many,
wrap_types=wrap_types,
normalize_for_parsing=False,
parsed_expression=parsed_expression,
param_fingerprint=param_fingerprint,
)
def _process_internal(
self,
sql: str,
parameters: "ParameterPayload",
config: "ParameterStyleConfig",
*,
dialect: str | None,
is_many: bool,
wrap_types: bool,
normalize_for_parsing: bool,
parsed_expression: Any = None,
param_fingerprint: Any | None = None,
) -> "ParameterProcessingResult":
cache_key = None
if self._cache_max_size > 0:
cache_key = self._make_processor_cache_key(
sql,
parameters,
config,
is_many,
dialect,
wrap_types,
normalize_for_parsing,
param_fingerprint=param_fingerprint,
)
cached_result = self._cache.get(cache_key)
if cached_result is not None:
self._cache.move_to_end(cache_key)
self._cache_hits += 1
# For static script compilation, parameters are embedded directly in SQL.
# Cache key includes parameter values, so a hit means same SQL with same values.
# Return None for parameters since the driver shouldn't receive any.
if config.needs_static_script_compilation:
return ParameterProcessingResult(
cached_result.sql,
None,
cached_result.parameter_profile,
sqlglot_sql=cached_result.sqlglot_sql,
parsed_expression=cached_result.parsed_expression,
input_named_parameters=cached_result.input_named_parameters,
applied_wrap_types=cached_result.applied_wrap_types,
)
# Return cached SQL transformation with NEW parameters transformed
# to match the cached SQL's placeholder format
transformed_params = self.transform_cached_parameters(
parameters,
cached_result.parameter_profile,
config,
input_named_parameters=cached_result.input_named_parameters,
is_many=is_many,
apply_wrap_types=cached_result.applied_wrap_types,
)
# Apply output transformer if present (it may further transform params)
final_sql = cached_result.sql
if config.output_transformer:
final_sql, transformed_params = config.output_transformer(final_sql, transformed_params)
return ParameterProcessingResult(
final_sql,
transformed_params,
cached_result.parameter_profile,
sqlglot_sql=cached_result.sqlglot_sql,
parsed_expression=cached_result.parsed_expression,
input_named_parameters=cached_result.input_named_parameters,
applied_wrap_types=cached_result.applied_wrap_types,
)
self._cache_misses += 1
param_info = self._validator.extract_parameters(sql)
original_styles = {p.style for p in param_info} if param_info else set()
needs_execution_conversion = self._needs_placeholder_conversion(param_info, config)
input_named_parameters = _named_parameters_for_style(param_info, config.default_execution_parameter_style)
if len(original_styles) > 1 and _is_sequence_payload(parameters, is_many):
ambiguous = ambiguous_index_placeholder(param_info)
if ambiguous is not None:
raise SQLSpecError(ambiguous_index_message(ambiguous))
has_qmark_escape = "??" in sql
unescape_output = has_qmark_escape and not executes_qmark(config)
if config.needs_static_script_compilation and param_info and not is_many:
return self._compile_static_script(
sql,
parameters,
config,
is_many,
cache_key,
dialect,
input_named_parameters=input_named_parameters,
unescape_output=unescape_output,
)
requires_mapping = self._needs_mapping_normalization(parameters, param_info, is_many)
if not requires_mapping and not needs_execution_conversion:
requires_mapping = _needs_named_binding(parameters, param_info, is_many)
if (
not needs_execution_conversion
and not config.type_coercion_map
and not config.output_transformer
and not requires_mapping
):
normalized_sql = self._normalize_sql_for_parsing(sql, param_info, config) if normalize_for_parsing else sql
fast_sql, fast_info = sql, param_info
if has_qmark_escape:
normalized_sql = unescape_qmark_operator(normalized_sql)[0]
if unescape_output:
fast_sql, fast_info = unescape_qmark_operator(sql, param_info)
result = ParameterProcessingResult(
fast_sql,
parameters,
ParameterProfile(fast_info),
sqlglot_sql=normalized_sql,
parsed_expression=parsed_expression,
input_named_parameters=input_named_parameters,
applied_wrap_types=False,
)
return self._store_cached_result(cache_key, result)
processed_sql, processed_parameters = sql, parameters
if requires_mapping:
target_style = self._select_execution_style(original_styles, config)
input_named_parameters = _named_parameters_for_style(param_info, target_style)
mapping_plan = self._converter._build_conversion_plan( # pyright: ignore[reportPrivateUsage]
param_info, target_style
)
processed_sql, processed_parameters, param_info = self._converter._convert_with_metadata(
processed_sql,
processed_parameters,
target_style,
is_many,
strict_named_parameters=config.strict_named_parameters,
param_info=param_info,
precomputed_plan=mapping_plan,
)
original_styles = {target_style}
needs_execution_conversion = False
if needs_execution_conversion:
target_style = self._select_execution_style(original_styles, config)
input_named_parameters = _named_parameters_for_style(param_info, target_style)
applied_wrap_types = False
if processed_parameters and wrap_types:
wrapped_parameters = self._wrap_parameter_types(processed_parameters)
if wrapped_parameters is not processed_parameters:
processed_parameters = wrapped_parameters
applied_wrap_types = True
if config.type_coercion_map and processed_parameters:
processed_parameters = self._coerce_parameter_types(processed_parameters, config.type_coercion_map, is_many)
processed_sql, processed_parameters, converted_param_info = self._execution_placeholders(
processed_sql,
processed_parameters,
config,
param_info,
original_styles,
needs_execution_conversion,
is_many,
)
if config.output_transformer:
processed_sql, processed_parameters = config.output_transformer(processed_sql, processed_parameters)
final_param_info = converted_param_info if converted_param_info is not None else param_info
sqlglot_sql = (
self._normalize_sql_for_parsing(processed_sql, final_param_info, config)
if normalize_for_parsing
else processed_sql
)
if has_qmark_escape:
sqlglot_sql = unescape_qmark_operator(sqlglot_sql)[0]
if unescape_output:
processed_sql, final_param_info = unescape_qmark_operator(processed_sql, final_param_info)
final_profile = ParameterProfile(final_param_info)
result = ParameterProcessingResult(
processed_sql,
processed_parameters,
final_profile,
sqlglot_sql=sqlglot_sql,
parsed_expression=parsed_expression,
input_named_parameters=input_named_parameters,
applied_wrap_types=applied_wrap_types,
)
return self._store_cached_result(cache_key, result)
def _compile_static_script(
self,
sql: str,
parameters: "ParameterPayload",
config: "ParameterStyleConfig",
is_many: bool,
cache_key: Any | None,
dialect: "str | None",
input_named_parameters: "tuple[str, ...]",
unescape_output: bool = False,
) -> "ParameterProcessingResult":
coerced_params = parameters
if config.type_coercion_map and parameters:
coerced_params = self._coerce_parameter_types(parameters, config.type_coercion_map, is_many)
static_sql, static_params = self._converter.convert_placeholder_style(
sql,
coerced_params,
ParameterStyle.STATIC,
is_many,
dialect=dialect,
strict_named_parameters=config.strict_named_parameters,
)
parse_sql = unescape_qmark_operator(static_sql)[0]
if unescape_output:
static_sql = parse_sql
result = ParameterProcessingResult(
static_sql,
static_params,
ParameterProfile.empty(),
sqlglot_sql=parse_sql,
input_named_parameters=input_named_parameters,
applied_wrap_types=False,
)
return self._store_cached_result(cache_key, result)
def _select_execution_style(
self, original_styles: "set[ParameterStyle]", config: "ParameterStyleConfig"
) -> "ParameterStyle":
if len(original_styles) == 1 and config.supported_execution_parameter_styles is not None:
original_style = next(iter(original_styles))
if original_style in config.supported_execution_parameter_styles:
return original_style
return config.default_execution_parameter_style or config.default_parameter_style
def _wrap_parameter_types(self, parameters: "ParameterPayload") -> "ConvertedParameters":
# Fast type dispatch for common types
param_type = type(parameters)
if param_type is list:
source = cast("list[Any]", parameters)
wrapped_values: list[Any] | None = None
for idx, value in enumerate(source):
wrapped = wrap_with_type(value)
if wrapped_values is None:
if wrapped is value:
continue
wrapped_values = source[:idx]
wrapped_values.append(wrapped)
if wrapped_values is None:
return parameters
return wrapped_values
if param_type is tuple:
tuple_source = cast("tuple[Any, ...]", parameters)
wrapped_values = [wrap_with_type(value) for value in tuple_source]
if all(wrapped is value for wrapped, value in zip(wrapped_values, tuple_source, strict=False)):
return parameters
return wrapped_values
if param_type is dict:
source_mapping = cast("dict[str, Any]", parameters)
wrapped_mapping: dict[str, Any] | None = None
for key, value in source_mapping.items():
wrapped = wrap_with_type(value)
if wrapped_mapping is None:
if wrapped is value:
continue
wrapped_mapping = dict(source_mapping)
wrapped_mapping[key] = wrapped
if wrapped_mapping is None:
return parameters
return wrapped_mapping
# Fallback to ABC checks for custom types
if isinstance(parameters, Sequence) and not isinstance(parameters, (str, bytes)):
wrapped_values = [wrap_with_type(value) for value in parameters]
if all(wrapped is value for wrapped, value in zip(wrapped_values, parameters, strict=False)):
return cast("ConvertedParameters", parameters)
return wrapped_values
if isinstance(parameters, Mapping):
fallback_mapping: dict[str, Any] | None = None
for key, value in parameters.items():
wrapped = wrap_with_type(value)
if fallback_mapping is None:
if wrapped is value:
continue
fallback_mapping = dict(parameters)
fallback_mapping[key] = wrapped
if fallback_mapping is None:
return cast("ConvertedParameters", parameters)
return fallback_mapping
return None
def _coerce_parameter_types(
self,
parameters: "ParameterPayload",
type_coercion_map: "dict[type, Callable[[Any], Any]]",
is_many: bool = False,
) -> "ConvertedParameters":
fallback_items = type_coercion_fallbacks(type_coercion_map)
return _coerce_parameters_payload(parameters, type_coercion_map, fallback_items, is_many)
def _store_cached_result(
self, cache_key: Any | None, result: "ParameterProcessingResult"
) -> "ParameterProcessingResult":
if self._cache_max_size <= 0 or cache_key is None:
return result
self._cache[cache_key] = result
self._cache.move_to_end(cache_key)
if len(self._cache) > self._cache_max_size:
self._cache.popitem(last=False)
return result
def _map_positional_to_named(
self,
parameters: "ConvertedParameters",
cached_profile: "ParameterProfile",
is_many: bool,
index_order: "tuple[int, ...]" = (),
) -> "ConvertedParameters":
"""Map a positional sequence to a dict keyed by cached placeholder names."""
cached_param_info = cached_profile.parameters
if not cached_param_info:
return parameters
if index_order:
cached_names = tuple(dict.fromkeys(param.name or f"param_{param.ordinal}" for param in cached_param_info))
if len(cached_names) == len(index_order):
if is_many and isinstance(parameters, (list, tuple)):
return [
_index_row_to_named(row, cached_names, index_order) if isinstance(row, (list, tuple)) else row
for row in cast("Sequence[Any]", parameters)
]
if isinstance(parameters, (list, tuple)):
return _index_row_to_named(parameters, cached_names, index_order)
if is_many and isinstance(parameters, (list, tuple)):
rows = cast("Sequence[Any]", parameters)
mapped_rows: list[Any] = []
for row in rows:
if isinstance(row, Mapping):
mapped_rows.append(_align_cached_mapping(row, cached_param_info))
continue
if isinstance(row, (list, tuple)) and not isinstance(row, (str, bytes, bytearray)):
mapped_rows.append(_sequence_row_to_named(row, cached_param_info))
continue
mapped_rows.append(row)
return mapped_rows
if isinstance(parameters, Mapping):
return _align_cached_mapping(parameters, cached_param_info)
if isinstance(parameters, (list, tuple)) and not isinstance(parameters, (str, bytes, bytearray)):
return _sequence_row_to_named(cast("Sequence[Any]", parameters), cached_param_info)
return parameters
def _map_named_to_positional(
self,
parameters: "ConvertedParameters",
named_order: "tuple[str, ...]",
is_many: bool,
strict: bool = False,
index_order: "tuple[int, ...]" = (),
) -> "ConvertedParameters":
"""Map named parameters (dict) to positional (tuple) using cached order.
Args:
parameters: Current parameters (dict or sequence).
named_order: Tuple of parameter names in placeholder order.
is_many: Whether this is execute_many.
strict: Whether to raise an error if required parameters are missing.
index_order: Source sequence slots in output occurrence order.
Returns:
Parameters converted to positional tuple if input was dict, else unchanged.
Raises:
SQLSpecError: If strict is True and required parameters are missing.
"""
if not named_order:
return parameters
param_type = type(parameters)
if is_many and (param_type is list or param_type is tuple):
parameter_rows = cast("Sequence[Any]", parameters)
updated_rows: list[Any] | None = None
for idx, row in enumerate(parameter_rows):
if type(row) is dict or isinstance(row, Mapping):
mapped_row: Any = _named_row_to_positional(row, named_order, strict)
elif index_order and isinstance(row, (list, tuple)):
mapped_row = _expand_index_row(row, index_order)
else:
mapped_row = row
if updated_rows is None:
if mapped_row is row:
continue
updated_rows = list(parameter_rows[:idx])
updated_rows.append(mapped_row)
if updated_rows is None:
return parameters
if param_type is tuple:
return tuple(updated_rows)
return updated_rows
if isinstance(parameters, Mapping):
return _named_row_to_positional(parameters, named_order, strict)
if index_order and isinstance(parameters, (list, tuple)):
return _expand_index_row(parameters, index_order)
return parameters
def _needs_mapping_normalization(
self, payload: "ParameterPayload", param_info: "list[ParameterInfo]", is_many: bool
) -> bool:
if not payload or not param_info:
return False
has_named_placeholders = any(param.style in _NAMED_STYLES for param in param_info)
if has_named_placeholders:
return False
if not is_many:
return False
# Fast type dispatch for common types
payload_type = type(payload)
if payload_type is dict:
return True
if payload_type is list or payload_type is tuple:
# Check if any item is a dict (fast path) or Mapping (fallback)
seq_payload = cast("Sequence[Any]", payload)
for item in seq_payload:
item_type = type(item)
if item_type is dict:
return True
if isinstance(item, Mapping):
return True
return False
# Fallback for custom types
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
def _normalize_sql_for_parsing(
self, sql: str, param_info: "list[ParameterInfo]", config: "ParameterStyleConfig"
) -> str:
"""Normalize SQL for sqlglot parsing by converting unsupported parameter styles.
When a parameter style is not in config.supported_parameter_styles (what sqlglot
can parse for this dialect), convert it to config.default_parameter_style.
Args:
sql: SQL string with parameters.
param_info: List of detected parameter placeholders.
config: Parameter style configuration.
Returns:
SQL string with parameters converted to a sqlglot-compatible style.
"""
if not self._needs_parse_normalization(param_info, config):
return sql
# Convert to the default style that sqlglot can parse for this dialect
target_style = config.default_parameter_style
normalized_sql, _ = self._converter.convert_placeholder_style(
sql, None, target_style, is_many=False, param_info=param_info
)
return normalized_sql
def _make_processor_cache_key(
self,
sql: str,
parameters: "ParameterPayload",
config: "ParameterStyleConfig",
is_many: bool,
dialect: str | None,
wrap_types: bool,
normalize_for_parsing: bool,
*,
param_fingerprint: Any | None = None,
) -> tuple[Any, ...]:
if param_fingerprint is None:
# For static script compilation, we must include actual values in the fingerprint
if config.needs_static_script_compilation:
param_fingerprint = value_fingerprint(parameters)
else:
# Use structural fingerprint (keys + types, not values) for better cache hit rates
param_fingerprint = structural_fingerprint(parameters, is_many)
dialect_marker = dialect or "default"
# Include both input and execution parameter styles to avoid cache collisions
input_style = config.default_parameter_style.value if config.default_parameter_style else "unknown"
exec_style = (
config.default_execution_parameter_style.value if config.default_execution_parameter_style else input_style
)
return _make_cache_key_tuple(
sql, param_fingerprint, input_style, exec_style, dialect_marker, is_many, wrap_types, normalize_for_parsing
)
def _needs_placeholder_conversion(self, param_info: "list[ParameterInfo]", config: "ParameterStyleConfig") -> bool:
"""Determine whether execution placeholder conversion is required."""
if config.needs_static_script_compilation:
return True
if not param_info:
return False
current_styles = {param.style for param in param_info}
if (
config.allow_mixed_parameter_styles
and len(current_styles) > 1
and config.supported_execution_parameter_styles is not None
and len(config.supported_execution_parameter_styles) > 1
and all(style in config.supported_execution_parameter_styles for style in current_styles)
):
return False
if len(current_styles) > 1:
return True
if len(current_styles) == 1:
current_style = next(iter(current_styles))
supported_styles = config.supported_execution_parameter_styles
if supported_styles is None:
return True
return current_style not in supported_styles
return True
def _needs_parse_normalization(self, param_info: "list[ParameterInfo]", config: "ParameterStyleConfig") -> bool:
"""Check if SQL needs normalization before sqlglot parsing.
A style needs normalization if it's NOT in config.supported_parameter_styles,
which represents what sqlglot can parse for this driver's dialect.
Args:
param_info: List of detected parameter placeholders.
config: Parameter style configuration with supported_parameter_styles.
Returns:
True if any parameter style is not supported by sqlglot for this dialect.
"""
supported = config.supported_parameter_styles
return any(p.style not in supported for p in param_info)
def _execution_placeholders(
self,
sql: str,
parameters: "ParameterPayload",
config: "ParameterStyleConfig",
param_info: "list[ParameterInfo]",
original_styles: "set[ParameterStyle]",
needs_execution_conversion: bool,
is_many: bool,
) -> "tuple[str, ConvertedParameters, list[ParameterInfo] | None]":
if not needs_execution_conversion:
# Convert parameters to concrete type for return
if parameters is None:
return sql, None, None
if isinstance(parameters, (dict, list, tuple)):
return sql, parameters, None
if isinstance(parameters, Mapping):
return sql, dict(parameters), None
if isinstance(parameters, Sequence) and not isinstance(parameters, (str, bytes)):
return sql, list(parameters), None
if len(param_info) == 1:
return sql, [parameters], None
return sql, None, None
target_style = self._select_execution_style(original_styles, config)
execution_plan = self._converter._build_conversion_plan( # pyright: ignore[reportPrivateUsage]
param_info, target_style
)
preserve_batch = (
is_many
and config.preserve_original_params_for_many
and isinstance(parameters, (list, tuple))
and original_styles == {target_style}
)
processed_sql, processed_parameters, converted_param_info = self._converter._convert_with_metadata(
sql,
parameters,
target_style,
is_many,
strict_named_parameters=config.strict_named_parameters,
param_info=param_info,
precomputed_plan=execution_plan,
preserve_original_batch=preserve_batch,
)
return processed_sql, processed_parameters, converted_param_info
def type_coercion_dispatcher(
fallback_items: "tuple[TypeCoercionFallback, ...]",
) -> "TypeDispatcher[Callable[[Any], Any]]":
"""Return the process-wide dispatcher registered for the fallback items."""
dispatcher = _TYPE_COERCION_DISPATCHERS.get(fallback_items)
if dispatcher is not None:
return dispatcher
dispatcher = TypeDispatcher["Callable[[Any], Any]"]()
dispatcher.register_all(fallback_items)
_TYPE_COERCION_DISPATCHERS[fallback_items] = dispatcher
return dispatcher
def type_coercion_fallbacks(
type_coercion_map: "dict[type, Callable[[Any], Any]] | None",
) -> "tuple[TypeCoercionFallback, ...]":
"""Return the ordered fallback items for a coercion map, empty when none is configured."""
if not type_coercion_map:
return ()
return tuple(type_coercion_map.items())
def apply_type_coercion(
value: object,
type_coercion_map: "dict[type, Callable[[Any], Any]]",
fallback_items: "tuple[TypeCoercionFallback, ...]",
) -> object:
"""Coerce a value by exact type first, then by the shared subclass/ABC dispatcher."""
value_type = type(value)
exact_converter = type_coercion_map.get(value_type)
if exact_converter is not None:
return exact_converter(value)
fallback_converter = type_coercion_dispatcher(fallback_items).get(value)
if fallback_converter is not None:
return fallback_converter(value)
return value
def _coerce_nested_value(
value: object,
type_coercion_map: "dict[type, Callable[[Any], Any]]",
fallback_items: "tuple[TypeCoercionFallback, ...]",
) -> object:
# Fast type dispatch for common types
value_type = type(value)
if value_type is list or value_type is tuple:
seq_value = cast("Sequence[Any]", value)
return [_coerce_parameter_value(item, type_coercion_map, fallback_items) for item in seq_value]
if value_type is dict:
dict_value = cast("dict[Any, Any]", value)
return {key: _coerce_parameter_value(val, type_coercion_map, fallback_items) for key, val in dict_value.items()}
return value
def _coerce_parameter_value(
value: object,
type_coercion_map: "dict[type, Callable[[Any], Any]]",
fallback_items: "tuple[TypeCoercionFallback, ...]",
) -> object:
if value is None:
return value
value_type = type(value)
# Fast path: check TypedParameter by type identity
if value_type is TypedParameter:
typed_param = cast("TypedParameter", value)
wrapped_value: object = typed_param.value
if wrapped_value is None:
return wrapped_value
coerced = apply_type_coercion(wrapped_value, type_coercion_map, fallback_items)
if coerced is wrapped_value:
return wrapped_value
return _coerce_nested_value(coerced, type_coercion_map, fallback_items)
coerced = apply_type_coercion(value, type_coercion_map, fallback_items)
if coerced is value:
return value
return _coerce_nested_value(coerced, type_coercion_map, fallback_items)
def _coerce_sequence_if_needed(
seq_value: "Sequence[Any]",
type_coercion_map: "dict[type, Callable[[Any], Any]]",
fallback_items: "tuple[TypeCoercionFallback, ...]",
) -> "Sequence[Any] | list[Any]":
updated_seq: list[Any] | None = None
for idx, item in enumerate(seq_value):
coerced_value = _coerce_parameter_value(item, type_coercion_map, fallback_items)
if updated_seq is None:
if coerced_value is item:
continue
updated_seq = list(seq_value[:idx])
updated_seq.append(coerced_value)
if updated_seq is None:
return seq_value
return updated_seq
def _coerce_mapping_if_needed(
mapping: "Mapping[Any, Any]",
type_coercion_map: "dict[type, Callable[[Any], Any]]",
fallback_items: "tuple[TypeCoercionFallback, ...]",
) -> "Mapping[Any, Any] | dict[Any, Any]":
updated_mapping: dict[Any, Any] | None = None
for key, val in mapping.items():
coerced_value = _coerce_parameter_value(val, type_coercion_map, fallback_items)
if updated_mapping is None:
if coerced_value is val:
continue
updated_mapping = dict(mapping)
updated_mapping[key] = coerced_value
if updated_mapping is None:
return mapping
return updated_mapping
def _coerce_parameter_set(
param_set: object,
type_coercion_map: "dict[type, Callable[[Any], Any]]",
fallback_items: "tuple[TypeCoercionFallback, ...]",
) -> object:
# Fast type dispatch for common types
param_type = type(param_set)
if param_type is list:
return _coerce_sequence_if_needed(cast("list[Any]", param_set), type_coercion_map, fallback_items)
if param_type is tuple:
seq_value = cast("tuple[Any, ...]", param_set)
coerced_seq = _coerce_sequence_if_needed(seq_value, type_coercion_map, fallback_items)
if coerced_seq is seq_value:
return seq_value
return tuple(cast("list[Any]", coerced_seq))
if param_type is dict:
return _coerce_mapping_if_needed(cast("dict[Any, Any]", param_set), type_coercion_map, fallback_items)
# Fallback to ABC checks for custom types
if isinstance(param_set, Sequence) and not isinstance(param_set, (str, bytes)):
seq_fallback = param_set
coerced_seq = _coerce_sequence_if_needed(seq_fallback, type_coercion_map, fallback_items)
if coerced_seq is seq_fallback:
return param_set
return coerced_seq
if isinstance(param_set, Mapping):
coerced_mapping = _coerce_mapping_if_needed(param_set, type_coercion_map, fallback_items)
if coerced_mapping is param_set:
return param_set
return coerced_mapping
return _coerce_parameter_value(param_set, type_coercion_map, fallback_items)
def _coerce_parameters_payload(
parameters: "ParameterPayload",
type_coercion_map: "dict[type, Callable[[Any], Any]]",
fallback_items: "tuple[TypeCoercionFallback, ...]",
is_many: bool,
) -> object:
# Fast type dispatch for common types
param_type = type(parameters)
if param_type is list:
seq_params = cast("list[Any]", parameters)
if is_many:
updated_many: list[Any] | None = None
for idx, param_set in enumerate(seq_params):
coerced_set = _coerce_parameter_set(param_set, type_coercion_map, fallback_items)
if updated_many is None:
if coerced_set is param_set:
continue
updated_many = seq_params[:idx]
updated_many.append(coerced_set)
if updated_many is None:
return seq_params
return updated_many
return _coerce_sequence_if_needed(seq_params, type_coercion_map, fallback_items)
if param_type is tuple:
tuple_params = cast("tuple[Any, ...]", parameters)
if is_many:
return [_coerce_parameter_set(param_set, type_coercion_map, fallback_items) for param_set in tuple_params]
coerced_tuple = _coerce_sequence_if_needed(tuple_params, type_coercion_map, fallback_items)
return list(coerced_tuple)
if param_type is dict:
dict_params = cast("dict[Any, Any]", parameters)
return _coerce_mapping_if_needed(dict_params, type_coercion_map, fallback_items)
# Fallback to ABC checks for custom types
if is_many and isinstance(parameters, Sequence) and not isinstance(parameters, (str, bytes)):
return [_coerce_parameter_set(param_set, type_coercion_map, fallback_items) for param_set in parameters]
if isinstance(parameters, Sequence) and not isinstance(parameters, (str, bytes)):
coerced_sequence = _coerce_sequence_if_needed(parameters, type_coercion_map, fallback_items)
return list(coerced_sequence)
if isinstance(parameters, Mapping):
coerced_mapping = _coerce_mapping_if_needed(parameters, type_coercion_map, fallback_items)
return dict(coerced_mapping)
return _coerce_parameter_value(parameters, type_coercion_map, fallback_items)
def _make_cache_key_tuple(
sql: str,
param_fingerprint: Any,
input_style: str,
exec_style: str,
dialect: str | None,
is_many: bool,
wrap_types: bool | None = None,
normalize_for_parsing: bool | None = None,
) -> tuple[Any, ...]:
"""Build the shared SQL and parameter processor cache key tuple."""
if wrap_types is None and normalize_for_parsing is None:
return (sql, param_fingerprint, input_style, exec_style, dialect, is_many)
return (sql, param_fingerprint, input_style, exec_style, dialect, is_many, wrap_types, normalize_for_parsing)
def _explicit_index_order(input_named_parameters: "tuple[str, ...]") -> "tuple[int, ...]":
"""Return zero-based indexes when every cached input name is a written ``$n`` / ``:n`` index."""
if not input_named_parameters:
return ()
indexes: list[int] = []
for name in input_named_parameters:
if not name.isdigit() or name == "0":
return ()
indexes.append(int(name) - 1)
return tuple(indexes)
def _validate_index_row(row: "Sequence[Any]", index_order: "tuple[int, ...]") -> None:
distinct = len(set(index_order))
if len(row) != distinct or max(index_order) >= len(row):
msg = (
f"Parameter count mismatch: {len(row)} parameters provided but "
f"{distinct} distinct positional indexes referenced."
)
raise SQLSpecError(msg)
def _expand_index_row(row: "Sequence[Any]", index_order: "tuple[int, ...]") -> "list[Any] | tuple[Any, ...]":
_validate_index_row(row, index_order)
expanded = [row[index] for index in index_order]
return expanded if isinstance(row, list) else tuple(expanded)
def _index_row_to_named(
row: "Sequence[Any]", cached_names: "tuple[str, ...]", index_order: "tuple[int, ...]"
) -> "dict[str, Any]":
_validate_index_row(row, index_order)
return {name: row[index] for name, index in zip(cached_names, index_order, strict=True)}
def _is_generated_name(name: str) -> bool:
return name.isdigit() or (name.startswith("param_") and name[6:].rstrip("_p").isdigit())
def _named_row_to_positional(
row: "Mapping[str, Any]", named_order: "tuple[str, ...]", strict: bool
) -> "tuple[Any, ...]":
"""Resolve cached placeholder names from a mapping without binding named values positionally."""
generated = [name for name in dict.fromkeys(named_order) if name not in row and _is_generated_name(name)]
if not generated:
if strict:
_validate_missing_parameters(named_order, row)
return tuple(row.get(name) for name in named_order)
reserved = set(named_order)
spare_keys = [key for key in row if key not in reserved]
all_indexes = all(name.isdigit() for name in named_order)
resolved: dict[str, Any] = {}
for rank, name in enumerate(generated):
position = int(name) - 1 if all_indexes else rank
if 0 <= position < len(spare_keys):
resolved[name] = row[spare_keys[position]]
missing = [name for name in named_order if name not in row and name not in resolved]
if missing and strict:
msg = f"Missing required parameters: {sorted(set(missing))}"
raise SQLSpecError(msg)
return tuple(row[name] if name in row else resolved.get(name) for name in named_order)
def _generated_name_order(name: str) -> int:
return int(name[6:].rstrip("_p")) if name.startswith("param_") else int(name)
def _align_cached_mapping(
parameters: "Mapping[str, Any]", cached_param_info: "Sequence[ParameterInfo]"
) -> "Mapping[str, Any]":
names = tuple(dict.fromkeys(param.name for param in cached_param_info if param.name is not None))
generated = [name for name in names if name not in parameters and _is_generated_name(name)]
if not generated:
return parameters
reserved = set(names)
spare_keys = [key for key in parameters if key not in reserved]
generated.sort(key=_generated_name_order)
if len(spare_keys) < len(generated):
return parameters
aligned = {name: parameters[name] for name in names if name in parameters}
for name, key in zip(generated, spare_keys, strict=False):
aligned[name] = parameters[key]
return aligned
def _named_parameters_for_style(
param_info: "list[ParameterInfo]", target_style: "ParameterStyle | None"
) -> "tuple[str, ...]":
has_named = any(p.style in _NAMED_STYLES for p in param_info)
if has_named and any(p.name is None for p in param_info):
generated_names = _named_parameter_names(param_info)
names = tuple(p.name if p.name is not None else generated_names[_parameter_lookup_key(p)] for p in param_info)
else:
names = tuple(p.name for p in param_info if p.name is not None)
if target_style in _EXPANDING_POSITIONAL_STYLES:
return names
unique_names = tuple(dict.fromkeys(names))
if target_style in {ParameterStyle.NUMERIC, ParameterStyle.POSITIONAL_COLON} and all(
param.style in {ParameterStyle.NUMERIC, ParameterStyle.POSITIONAL_COLON} for param in param_info
):
return tuple(sorted(unique_names, key=int))
return unique_names
def _validate_missing_parameters(named_order: Sequence[str], parameters: Mapping[str, Any]) -> None:
"""Validate that all required parameters are present in parameters mapping.
Args:
named_order: Expected parameter names in order.
parameters: Parameter mapping to validate.
Raises:
SQLSpecError: If any required parameters are missing.
"""
missing = [name for name in named_order if name not in parameters]
if missing:
msg = f"Missing required parameters: {missing}"
raise SQLSpecError(msg)
def _fingerprint_execute_many(parameters: "Sequence[Any]") -> Any:
"""Generate fingerprint for execute_many parameters.
Extracted to reduce code duplication and allow inlining of the common single-execution path.
"""
first = parameters[0]
first_type = type(first)
# Fast type dispatch for first element
if first_type is dict:
keys = tuple(first.keys())
type_sig = tuple(type(v) for v in first.values())
return ("many_dict", keys, type_sig)
if first_type is list or first_type is tuple:
return ("many_seq", tuple(type(v) for v in first))
# Fallback to ABC checks
if isinstance(first, Mapping):
keys = tuple(first.keys())
type_sig = tuple(type(v) for v in first.values())
return ("many_dict", keys, type_sig)
if isinstance(first, Sequence) and not isinstance(first, (str, bytes)):
return ("many_seq", tuple(type(v) for v in first))
# Scalar values in sequence for execute_many
return ("many_scalar", first_type)
def executes_qmark(config: "ParameterStyleConfig") -> bool:
"""Return True when the driver itself reads ``?`` as a placeholder."""
if config.default_execution_parameter_style == ParameterStyle.QMARK:
return True
supported = config.supported_execution_parameter_styles
return supported is not None and ParameterStyle.QMARK in supported
def _is_sequence_payload(parameters: "ParameterPayload", is_many: bool) -> bool:
if not parameters or isinstance(parameters, (str, bytes, bytearray, Mapping)):
return False
if not isinstance(parameters, Sequence):
return False
if is_many:
first = parameters[0]
return isinstance(first, Sequence) and not isinstance(first, (str, bytes, bytearray))
return True
def _needs_named_binding(parameters: "ParameterPayload", param_info: "list[ParameterInfo]", is_many: bool) -> bool:
"""Return True when a sequence payload must be keyed by natively supported named placeholders."""
if not param_info or not parameters:
return False
if _is_sequence_payload(parameters, is_many):
return all(param.style in _NAMED_STYLES for param in param_info)
if not is_many and isinstance(parameters, Mapping):
return all(param.style not in _NAMED_STYLES for param in param_info)
return False
def _sequence_row_to_named(row: "Sequence[Any]", cached_param_info: "Sequence[ParameterInfo]") -> "dict[str, Any]":
expected = len({param.name or f"param_{param.ordinal}" for param in cached_param_info})
if len(row) != expected:
msg = f"Parameter count mismatch: {len(row)} parameters provided but {expected} placeholders referenced."
raise SQLSpecError(msg)
named: dict[str, Any] = {}
for param in cached_param_info:
name = param.name or f"param_{param.ordinal}"
if name in named:
continue
slot = len(named)
if slot < len(row):
named[name] = row[slot]
return named
def _mapping_rows_to_positional(parameters: "ConvertedParameters", is_many: bool) -> "ConvertedParameters":
if is_many and isinstance(parameters, (list, tuple)):
return [tuple(row.values()) if isinstance(row, Mapping) else row for row in parameters]
if isinstance(parameters, Mapping):
return tuple(parameters.values())
return parameters
