8000 PGValues · sqlalchemy/sqlalchemy Wiki · GitHub
[go: up one dir, main page]
Skip to content

PGValues

Jump to bottom
mike bayer edited this page Oct 17, 2022 · 7 revisions

Render a SQL VALUES() construct within SELECT statements

Note - SQLAlchemy includes the SQL "VALUES" construct built in as of the 1.4 series:

https://docs.sqlalchemy.org/en/14/core/selectable.html?highlight=values#sqlalchemy.sql.expression.values

A modern 1.4/2.0 compatible version of the original example is as follows:

from sqlalchemy import Column
from sqlalchemy import create_engine
from sqlalchemy import Integer
from sqlalchemy import MetaData
from sqlalchemy import String
from sqlalchemy import values
from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import Session
from sqlalchemy.sql import column

Base = declarative_base()

m1 = MetaData()


class T(Base):
    __tablename__ = "mytable"

    mykey = Column(Integer, primary_key=True)
    mytext = Column(String(50))
    myint = Column(Integer)

t2 = values(
    column("mykey", Integer),
    column("mytext", String),
    column("myint", Integer),
    name="myvalues",
).data(
    [
        (1, "textA", 99),
        (2, "textB", 88),
    ]
)
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(
    {"mytext": t2.c.mytext, "myint": t2.c.myint},
    synchronize_session=False
)

LEGACY EXAMPLE - For the 1.3 series only.

This recipe produces a Postgresql VALUES construct.

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)

    @property
    def _from_objects(self):
        return [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)
    )

The query at the end renders:

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
Clone this wiki locally
0