python - 使用 SQLAlchemy 的 PostgreSQL 值列表(表构造函数)。如何进行原生SQL查询？

Question

我正在尝试使用 PostgreSQL 值列表 https://www.postgresql.org/docs/current/queries-values.html (也称为表构造函数)在 Python 中使用 SQLAlchemy。

这是在 PostgreSQL 中运行的 SQL

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

我构建了一个股票代码列表并将其作为参数传递。它由 SQLAlchecmy 转换为

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

如果将 list 用作 IN 子句的参数，看起来不错。但就我而言，这不起作用。如何正确提供参数列表？

这是我的代码:

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)

我很少看到类似VALUES clause in SQLAlchemy这样的相关讨论并检查了当前的解决方案，如 https://github.com/sqlalchemy/sqlalchemy/wiki/PGValues 。但是，我没有看到 native SQL 查询的解决方案。有吗？

Answer 1

对于一般情况，表值构造函数 (TVC) 值应该是元组列表，而不是标量值列表

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

我们可以从中构建 TVC(VALUES 构造)

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

现在获取查询的最简单方法就是简单

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

我们可以做什么

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',)]
    """

但是，如果您确实想要使用文字 SQL 查询，您也可以这样做

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',)]
    """

或者，如果您的 TVC 有很多列，并且您不想将它们全部键入，您可以让 SQLAlchemy 为您构建列列表

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)