如何在SQLAlchemy原生SQL查询中使用PostgreSQL VALUES列表

问题分析 #

你当前使用expanding=True绑定列表参数时，SQLAlchemy会将列表展开为同一行的多个值（如VALUES (%s, %s, %s)），但PostgreSQL的VALUES表构造器需要每个值作为独立行（如VALUES (%s), (%s), (%s)），因此需要调整参数绑定的方式。

解决方案 #

方案一：动态生成VALUES占位符 #

手动根据列表长度生成对应数量的独立占位符，每个占位符对应一个参数，确保生成符合要求的VALUES结构：

修改save_new方法如下：

def save_new(self, companies):
    tickers = ['A', 'B', 'C']
    # 生成每个值对应的占位符，格式为(:t0), (:t1), ...
    placeholders = ", ".join([f":t{i}" for i in range(len(tickers))])
    # 拼接完整SQL语句
    query_text = f'''
            SELECT input_values.ticker
            FROM (VALUES {placeholders}) as input_values(ticker)
            '''
    database_engine = self.__database_connection.get_engine()
    with database_engine.connect() as connection:
        query = text(query_text)
        # 构造参数字典，每个占位符对应一个ticker值
        params = {f"t{i}": ticker for i, ticker in enumerate(tickers)}
        result = connection.execute(query, params)
        new_tickers = [row[0] for row in result]
        logging.info(new_tickers)

方案二：使用PostgreSQL的unnest函数替代VALUES #

如果场景允许，可利用PostgreSQL的unnest函数将数组直接转为行，这种方式参数传递更简洁：

from sqlalchemy import ARRAY, String

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

方案三：混合SQLAlchemy表达式与原生SQL #

利用SQLAlchemy的values()构造器生成正确的VALUES子句，再嵌入到原生SQL中，自动处理参数绑定：

from sqlalchemy import values, column, String

def save_new(self, companies):
    tickers = ['A', 'B', 'C']
    database_engine = self.__database_connection.get_engine()
    # 构造VALUES子句的SQLAlchemy表达式
    vals_clause = values(column('ticker', String)).data([(t,) for t in tickers])
    # 编译表达式为SQL字符串和参数字典
    vals_sql, vals_params = vals_clause.compile(
        dialect=database_engine.dialect,
        compile_kwargs={"literal_binds": False}
    ).as_sql()
    # 拼接原生SQL
    query_text = f'''
            SELECT input_values.ticker
            FROM ({vals_sql}) as input_values(ticker)
            '''
    with database_engine.connect() as connection:
        query = text(query_text)
        result = connection.execute(query, vals_params)
        new_tickers = [row[0] for row in result]
        logging.info(new_tickers)

内容的提问来源于stack exchange，提问作者P_M