Function _extend_values_for_multiparams

lib/sqlalchemy/sql/crud.py:1564–1607

(
    compiler: SQLCompiler,
    stmt: ValuesBase,
    compile_state: DMLState,
    initial_values: Sequence[_CrudParamElementStr],
    _column_as_key: Callable[..., str],
    kw: Dict[str, Any],
)

def _extend_values_for_multiparams(
    compiler: SQLCompiler,
    stmt: ValuesBase,
    compile_state: DMLState,
    initial_values: Sequence[_CrudParamElementStr],
    _column_as_key: Callable[..., str],
    kw: Dict[str, Any],
) -> List[Sequence[_CrudParamElementStr]]:
    values_0 = initial_values
    values = [initial_values]

    has_visiting_cte = kw.get("visiting_cte") is not None
    mp = compile_state._multi_parameters
    assert mp is not None
    for i, row in enumerate(mp[1:]):
        extension: List[_CrudParamElementStr] = []

        row = {_column_as_key(key): v for key, v in row.items()}

        for col, col_expr, param, accumulated_names in values_0:
            if col.key in row:
                key = col.key

                if coercions._is_literal(row[key]):
                    new_param = _create_bind_param(
                        compiler,
                        col,
                        row[key],
                        name=("%s_m%d" % (col.key, i + 1)),
                        force_anonymous=has_visiting_cte,
                        **kw,
                    )
                else:
                    new_param = compiler.process(row[key].self_group(), **kw)
            else:
                new_param = _process_multiparam_default_bind(
                    compiler, stmt, col, i, kw
                )

            extension.append((col, col_expr, new_param, accumulated_names))

        values.append(extension)

    return values


def _get_stmt_parameter_tuples_params(