PostgreSQL Values list(table constructor)with SQLAlchemy.如何进行原生SQL查询？

xwbd5t1u 于 12个月前发布在 PostgreSQL

关注(0)|答案(1)|浏览(105)

我尝试在Python中使用PostgreSQL值列表https://www.postgresql.org/docs/current/queries-values.html（也称为表构造函数）与SQLAlchemy。
在这里，PostgreSQL中的SQL

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

我建立了一个tickers列表，并将其作为参数传递。它由SQLAlchecmy转换为

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

如果将列表用作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这样的当前解决方案。但是，我没有看到原生SQL查询的解决方案。有吗？

postgresql

来源：https://stackoverflow.com/questions/77295731/postgresql-values-list-table-constructor-with-sqlalchemy-how-to-for-native-sq

1条答案

按热度按时间

yb3bgrhw1#

对于一般情况，表值构造器（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)

赞(0）回复(0）举报 12个月前

我来回答

PostgreSQL Values list(table constructor)with SQLAlchemy.如何进行原生SQL查询？

1条答案

相关问题

热门标签

最新问答