python - SQLAlchemy 是否支持 postgres 的 UPDATE.....FROM 语法？

Question

我想用 SQLAlchemy 编写这样的查询:

UPDATE mytable
SET 
  mytext = myvalues.mytext,
  myint = myvalues.myint
FROM (
  VALUES
    (1, 'textA', 99),
    (2, 'textB', 88),
    ...
) AS myvalues (mykey, mytext, myint)
WHERE mytable.mykey = myvalues.mykey

ORM 本身就支持这种东西吗？还是我需要使用 session.execute() 来运行原始 SQL？

Answer 1

通过邮件列表中的@zzzeek:

我们执行 UPDATE..FROM，但是在其中获取 VALUES 需要一些额外的方法，因为我们现在没有内置该结构。它的别名部分也不是本地内置的，它在 AS 部分中命名列。食谱在这里:https://bitbucket.org/zzzeek/sqlalchemy/wiki/UsageRecipes/PGValues .在某些时候，我向某人展示了 AS 部分，但那部分不在那里……好吧，我刚刚更新了它，它还需要帮助才能在 UPDATE 上下文中工作。所以:

from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql.expression import FromClause

class values(FromClause):
    named_with_column = True

    def __init__(self, columns, *args, **kw):
        self._column_args = columns
        self.list = args
        self.alias_name = self.name = kw.pop('alias_name', None)

    def _populate_column_collection(self):
        for c in self._column_args:
            c._make_proxy(self)


@compiles(values)
def compile_values(element, compiler, asfrom=False, **kw):
    columns = element.columns
    v = "VALUES %s" % ", ".join(
        "(%s)" % ", ".join(
                compiler.render_literal_value(elem, column.type)
                for elem, column in zip(tup, columns))
        for tup in element.list
    )
    if asfrom:
        if element.alias_name:
            v = "(%s) AS %s (%s)" % (v, element.alias_name, (", ".join(c.name for c in element.columns)))
        else:
            v = "(%s)" % v
    return v

if __name__ == '__main__':
    from sqlalchemy import MetaData, create_engine, String, Integer, Table, Column
    from sqlalchemy.sql import column
    from sqlalchemy.orm import Session, mapper
    m1 = MetaData()
    class T(object):
        pass
    t1 = Table('mytable', m1, Column('mykey', Integer, primary_key=True),
                    Column('mytext', String),
                    Column('myint', Integer))
    mapper(T, t1)
    t2 = values(
            [
                column('mykey', Integer),
                column('mytext', String),
                column('myint', Integer)
            ],


            (1, 'textA', 99),
            (2, 'textB', 88),

            alias_name='myvalues'
        )
    e = create_engine("postgresql://scott:tiger@localhost/test", echo=True)
    m1.create_all(e)
    sess = Session(e)
    sess.query(T).filter(T.mykey==t2.c.mykey).\
            update(dict(mytext=t2.c.mytext, myint=t2.c.myint))