SQLAlchemy 中的 VALUES 子句

Question

有没有办法在 SQLAlchemy 中构建 Query 对象，相当于：

SELECT * FROM (VALUES (1, 2, 3)) AS sq;

根据我在文档中看到的，VALUES 子句仅在与 INSERT 一起使用时出现。

Answer 1

我不太确定，但您可以尝试以下选项：

SELECT * FROM (select 1 as col1 union select 2 as col1 union select 3 as col1) AS sq

Answer 2

插入中的“VALUES”是标准 SQL，独立的“VALUES”关键字是 Postgresql 的东西。在PGValues 有一个快速编译器配方（复制到这里，以防有一天我更改 wiki）：

from sqlalchemy import *
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql.expression import FromClause
from sqlalchemy.sql import table, column

class values(FromClause):
    def __init__(self, *args):
        self.list = args

    def _populate_column_collection(self):
        self._columns.update(
            [("column%d" % i, column("column%d" % i))
                    for i in xrange(1, len(self.list[0]) + 1)]
        )

@compiles(values)
def compile_values(element, compiler, asfrom=False, **kw):
    v = "VALUES %s" % ", ".join(
        "(%s)" % ", ".join(compiler.render_literal_value(elem, None) for elem in tup)
        for tup in element.list
    )
    if asfrom:
        v = "(%s)" % v
    return v

if __name__ == '__main__':
    t1 = table('t1', column('a'), column('b'))
    t2 = values((1, 0.5), (2, -0.5)).alias('weights')
    print select([t1, t2]).select_from(t1.join(t2, t1.c.a==t2.c.column2))

Answer 3

我建议我的 zzzeek 的 sn-p 版本

• 将None 渲染为NULL
• 根据column类型指定字面量类型
• 处理 ARRAY 字面量（对 PostgreSQL 有用）

How to make use of bindparam() in a custom Compiled expression?

Answer 4

from sqlalchemy import select, func

select(['*']).select_from(func.values([1, 2, 3, 4]))

Answer 5

不是最好的解决方案，但它对我有用：

import sqlalchemy as sa

query = sa.select(['*']).select_from(sa.text("(VALUES (1,2,3)) as sq"))
connection.execute(query).fetchall()

Output: [(1, 2, 3)]

PS： VALUES 带有别名列示例：

import sqlalchemy as sa

query_cte = (
    sa.select([sa.column('age'), sa.column('name')])
        .select_from(sa.text("(VALUES (22, 'Bob'), (30, 'Julia')) as t (age, name)"))
).cte()
query_name = sa.select([query_cte.c.name])
connection.execute(query_name).fetchall()

Output: [('Bob',), ('Julia',)]

警告： 此解决方案适用于简单值。小心特殊的 符号或单词，它们应该被正确地转义。

Answer 6

我更进一步，实现了SELECT INTO <table> VALUES (...)。

以下实现旨在与 PostgreSQL 和 Python 3 一起使用，并处理不同类型的值（布尔、整数、浮点、JSON 和 varchar）：

import json
from sqlalchemy import exc
from sqlalchemy.dialects import postgresql
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql import sqltypes, FromClause, Select

class SelectInto(Select):
    def __init__(self, columns, into, *arg, **kw):
        super(SelectInto, self).__init__(columns, *arg, **kw)
        self.into = into

@compiles(SelectInto)
def _select_into(element, compiler, **kw):
    text = compiler.visit_select(element, **kw)
    text = text.replace("FROM", f"INTO {element.into} \nFROM")
    return text

class Values(FromClause):
    text_type = sqltypes.UnicodeText

    def __init__(self, cols, types=None, *args):
        self.cols = cols
        self.vals = args

        if isinstance(self.cols, str):
            self.cols = [c.strip().join('""') for c in self.cols.split(",")]

        if not types:
            self.types = [self.text_type for _ in range(len(self.cols))]
        elif len(cols) == len(types):
            self.types = [self._map_col_type(t) for t in types]
        else:
            raise exc.ArgumentError("Types do not correspond to columns")

    def _map_col_type(self, col_type):
        if isinstance(col_type, sqltypes.TypeEngine):
            return col_type

        col_type = col_type.lower()

        if col_type in ("bool", "boolean"):
            return sqltypes.Boolean
        elif col_type in ("int", "integer", "number"):
            return sqltypes.Integer
        elif col_type in ("float", "double"):
            return sqltypes.Float
        elif col_type in ("json",):
            return postgresql.json.JSON
        elif col_type in ("jsonb",):
            return postgresql.json.JSONB

        return self.text_type

@compiles(Values)
def _compile_values(element, compiler, **kw):
    value_cols = ",".join(element.cols)
    value_sets = ", ".join(
        "({values})".format(
            values=",".join(_compile_value(compiler, val, element.types[idx]) for idx, val in enumerate(tup))
        )
        for tup in element.vals
    )
    return f'(VALUES {value_sets}) AS "values" ({value_cols})'

def _compile_value(compiler, value, type_):
    if value is None:
        return "NULL"
    elif issubclass(type_, sqltypes.JSON):
        if isinstance(value, dict):
            value = json.dumps(value)
        return f"'{value}'::{type_.__name__}"

    if issubclass(type_, sqltypes.String):
        value = str(value)

    return compiler.render_literal_value(value, type_())

对此进行测试：

from sqlalchemy.sql.expression column

select_cols = [column(c) for c in 'one,two,three'.split(',')]
select_from = Values(['one', 'two', 'three'], ['varchar', 'int', 'bool'], *(('a',1,0),('aa',11,1),('aaa',111,0)))
select_into = SelectInto(select_cols, 'test_select').select_from(select_from)
print(select_into)

Answer 7

这现在在 SQLAlchemy 中原生可用。

你的例子可以写成：

from sqlalchemy import select, column, Integer
from sqlalchemy.sql import Values

select(Values(column('Number', Integer), name='sq').data([(1,), (2,), (3,)]))

似乎没有这方面的任何文档，但你可以看看测试用例https://github.com/sqlalchemy/sqlalchemy/blob/master/test/sql/test_values.py