How to pass a list of tuples for a VALUES clause?

Question

I am trying to run a query with a CTE of the following desired format:

SELECT * FROM (
  VALUES
      (1, 'A'),
      (2, 'B')
) temp_table(column_a, column_b)

Using SQLalchemy, I want to dynamically populate the data (1, 'A', etc.) in this table. I have tried the following:

import sqlalchemy as sa

q =  """
SELECT * FROM (
  VALUES
      (:column_a, :column_b)
) temp_table(column_a, column_b)
"""

params = [
    {'column_a': 1, 'column_b': 'A'},
    {'column_a': 2, 'column_b': 'B'}
]
conn.execute(sa.text(q), parameters=params)

However, because I am providing a list of parameters, this uses execute_many under the hood (as I understand it) and ends up running the query twice, with the first query being eg:

SELECT * FROM (
  VALUES
      (1, 'A')
) temp_table(column_a, column_b)

How do I properly parametrize this so that all of my tuples are inserted into the query in one go?

Answer 1

You can use the .values() function to construct the VALUES clause for you:

import sqlalchemy as sa

engine = sa.create_engine(
    "mssql+pyodbc://scott:tiger^5HHH@mssql_199"
)

values_data = [(1, "A"), (2, "B")]
values_clause = sa.values(
    sa.column("column_a", sa.Integer),
    sa.column("column_b", sa.String),
    name="temp_table",
).data(values_data)

statement = sa.select(sa.text("*")).select_from(values_clause)
engine.echo = True
with engine.begin() as conn:
    results = conn.execute(statement).all()
    """SQL emitted
    SELECT *
    FROM (VALUES (?, ?), (?, ?)) AS temp_table (column_a, column_b)
    [no key 0.00031s] (1, 'A', 2, 'B')
    """
    print(results)
    # [(1, 'A'), (2, 'B')]

(Edit)

You could also compile the VALUES clause with literal_binds and then embed that into SQL text:

cte_body = (
    sa.select(sa.text("*"))
    .select_from(values_clause)
    .compile(engine, compile_kwargs={"literal_binds": True})
)
statement = f"""\
WITH cte1 AS (
{cte_body}
)
SELECT * FROM cte1 
"""

engine.echo = True
with engine.begin() as conn:
    results = conn.execute(sa.text(statement)).all()
    """SQL emitted
    WITH cte1 AS (
    SELECT * 
    FROM (VALUES (1, 'A'), (2, 'B')) AS temp_table (column_a, column_b)
    )
    SELECT * FROM cte1 
    """
    print(results)
    # [(1, 'A'), (2, 'B')]