textasfrom compatibility with columns, query

[Mike Bayer]

    def test_via_textasfrom(self):
        User = self.classes.User
        s = create_session()

        eq_(
            s.query(User).from_statement(
                text("select * from users order by id").\
                        columns(id=Integer, name=String)
            ).all(),
            [User(id=8), User(id=9), User(id=10)](User(id=7),)
        )

[Mike Bayer]

29de1077b35fb264f4b172

[Mike Bayer]

a few more regressions before we had textasfrom:

diff --git a/test/sql/test_query.py b/test/sql/test_query.py
index 6e22276..c200878 100644
--- a/test/sql/test_query.py
+++ b/test/sql/test_query.py
@@ -1831,6 +1831,36 @@ class KeyTargetingTest(fixtures.TablesTest):
         assert stmt.c.a in row
         assert stmt.c.b in row

+    def test_columnclause_schema_column_four(self):
+        keyed2 = self.tables.keyed2
+
+        # this is also addressed by #2932
+
+        a, b = sql.column('keyed2_a'), sql.column('keyed2_b')
+        stmt = text("select a AS keyed2_a, b AS keyed2_b from keyed2").columns(a, b)
+        row = testing.db.execute(stmt).first()
+
+        assert keyed2.c.a in row
+        assert keyed2.c.b in row
+        assert a in row
+        assert b in row
+        assert stmt.c.keyed2_a in row
+        assert stmt.c.keyed2_b in row
+
+    def test_columnclause_schema_column_five(self):
+        keyed2 = self.tables.keyed2
+
+        # this is also addressed by #2932
+
+        stmt = text("select a AS keyed2_a, b AS keyed2_b from keyed2").columns(
+                            keyed2_a=CHAR, keyed2_b=CHAR)
+        row = testing.db.execute(stmt).first()
+
+        assert keyed2.c.a in row
+        assert keyed2.c.b in row
+        assert stmt.c.keyed2_a in row
+        assert stmt.c.keyed2_b in row
+


 class LimitTest(fixtures.TestBase):

that is, the ColumnClause is directly named that of a column._label.

[Mike Bayer]

5c188f6c1ce85eaace27f052

[Mike Bayer]

OK how about this:

from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.sql import column
Base = declarative_base()

class A(Base):
    __tablename__ = 'a'

    id = Column(Integer, primary_key=True)
    data = Column(String)


e = create_engine("sqlite://", echo=True)
Base.metadata.create_all(e)

sess = Session(e)
sess.add_all([A(data='a2')](A(data='a1'),))

sess.commit()

result = sess.query(A).from_statement(
                text("SELECT id AS x, data AS y FROM a").
                    columns(A.id.label("x"), A.data.label("y"))
                ).all()

to make that work we need to do this:

diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index d4b0807..ad81503 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -618,7 +618,7 @@ class SQLCompiler(Compiled):
         if populate_result_map:
             for c in taf.c:
                 self._add_to_result_map(
-                        c.key, c.key, (c,), c.type
+                        c.key, c.key, tuple(c.proxy_set), c.type
                 )

         text = self.process(taf.element, **kw)

so, if the text() is made against a column on an aliased() - does c.proxy_set cover too broad of an area? I think so so maybe instead we really want this:

diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index d4b0807..f2a8b3e 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -618,7 +618,7 @@ class SQLCompiler(Compiled):
         if populate_result_map:
             for c in taf.c:
                 self._add_to_result_map(
-                        c.key, c.key, (c,), c.type
+                        c.key, c.key, (c, ) + tuple(c._proxies), c.type
                 )

         text = self.process(taf.element, **kw)

we'd need to test a text() that is hitting "cls" plus "aliased(cls)".

[Mike Bayer]

the same behavior applies to select() too. this raises an error too, meaning from_statement(select()) is also using key_fallback and matching on names. That shouldn't be the case.

stmt = select([A.data.label('asdfdsa')](A.id.label('asdf'),))
result = sess.query(A).from_statement(stmt).all()

[Mike Bayer]

maybe column.make_proxy() should have a little more going on - not just _proxies, but also, "_lookup_equivalent", which is flagged for both a select() and a text() when it makes the initial column proxy. that gets added to the result_map. _lookup_equivalent gets erased for any proxies subsequent to that, e.g. such as aliased().

the fact that a select() is never embedded as a subquery without being an alias might work well with this concept.

[Mike Bayer]

bf934018a52b4fe4c43745f is as far as we can go. I tried also doing this:

--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -274,7 +274,7 @@ class _CompileLabel(visitors.Visitable):
     __slots__ = 'element', 'name'

     def __init__(self, col, name, alt_names=()):
-        self.element = col
+        self.element = self._element = col
         self.name = name
         self._alt_names = (col,) + alt_names

@@ -511,7 +511,7 @@ class SQLCompiler(Compiled):
                 add_to_result_map(
                         labelname,
                         label.name,
-                        (label, labelname, ) + label._alt_names,
+                        (label, labelname, label._element) + label._alt_names,
                         label.type
                 )

which allows the idea of text("select A as X").columns(column("A").label("X")), but we get into naming conflicts there, and test.sql.test_query's tests for label conflicts does find one. The issue is if there are multiple columns named "A" with different labels, we can't do it as above. It's possible we could do something specific for TextAsFrom, but it would have to correct for the same column() present twice perhaps. I don't think it's a need right now.