PostgreSQL Values list (table constructor) with SQLAlchemy. How to for native SQL query?

Question

I am trying to use PostgreSQL Values list https://www.postgresql.org/docs/current/queries-values.html (also known as table constructor) in Python with SQLAlchemy.

Here the SQL working in PostgreSQL

SELECT input_values.ticker
FROM (VALUES ('A'), ('B'), ('C')) as input_values(ticker)

I built a list of tickers and passing it as argument. It converted by SQLAlchecmy to

SELECT input_values.ticker
FROM (VALUES (%(input_values_1)s, %(input_values_2)s, %(input_values_3)s)) as input_values(ticker) 

Which looks good if would list be used as param with IN clause. But in my case this not works. How to provide parameters list correctly?

Here the code I have:

import logging

from injector import inject
from sqlalchemy import bindparam
from sqlalchemy.sql import text

from database.database_connection import DatabaseConnection


class CompaniesDAO:
    FIND_NEW_QUERY = '''
            SELECT input_values.ticker
            FROM (VALUES :input_values) as input_values(ticker)
            '''

    @inject
    def __init__(self, database_connection: DatabaseConnection):
        self.__database_connection = database_connection

    def save_new(self, companies):
        tickers = ['A', 'B', 'C']
        input_values = {'input_values': tickers}
        database_engine = self.__database_connection.get_engine()
        with database_engine.connect() as connection:
            query = text(CompaniesDAO.FIND_NEW_QUERY)
            query = query.bindparams(bindparam('input_values', expanding=True))
            result = connection.execute(query, input_values)
            new_tickers = [row[0] for row in result]
            logging.info(new_tickers)

I saw few related discussions like VALUES clause in SQLAlchemy and checked current solution like https://github.com/sqlalchemy/sqlalchemy/wiki/PGValues. However, I not see solution for native SQL query. Is there any?

Answer 1

For the general case, the Table Value Constructor (TVC) values should be a list of tuples, not a list of scalar values

tickers = [('A',), ('B',), ('C',)]

from which we can then build the TVC (VALUES construct)

tvc = values(
    column("ticker", String),
    name="input_values"
).data(tickers)

Now the easiest way to get the query is simply

qry = select(text("input_values.ticker")).select_from(tvc)

with which we can do

engine.echo = True
with engine.begin() as conn:
    results = conn.execute(qry).all()
    """
    SELECT input_values.ticker 
    FROM (VALUES (%(param_1)s), (%(param_2)s), (%(param_3)s)) AS input_values (ticker)
    [no key 0.00036s] {'param_1': 'A', 'param_2': 'B', 'param_3': 'C'}
    """
    print(results)
    """
    [('A',), ('B',), ('C',)]
    """

However, if you really want to use a literal SQL query you could also do

tvc = values(
    column("ticker", String),
    name="input_values",
    literal_binds=True,
).data(tickers)

qry = text(
    "SELECT input_values.ticker "
    f"FROM ({tvc.compile(engine)}) AS input_values (ticker)"
)

engine.echo = True
with engine.begin() as conn:
    results = conn.execute(qry).all()
    """
    SELECT input_values.ticker FROM (VALUES ('A'), ('B'), ('C')) AS input_values (ticker)
    [generated in 0.00026s] {}
    """
    print(results)
    """
    [('A',), ('B',), ('C',)]
    """

Or if your TVC had a lot of columns and you didn't want to type them all out you could let SQLAlchemy build the column list for you

tvc = values(
    column("ticker", String),
    name="input_values",
    literal_binds=True,
).data(tickers)

basic_select = select(text("*")).select_from(tvc)
tvc_compiled = str(basic_select.compile(engine))[15:]
print(tvc_compiled)
# (VALUES ('A'), ('B'), ('C')) AS input_values (ticker)