pep484 ORM / SQL result support

after some experimentation it seems mypy is more amenable
to the generic types being fully integrated rather than
having separate spin-off types.   so key structures
like Result, Row, Select become generic.  For DML
Insert, Update, Delete, these are spun into type-specific
subclasses ReturningInsert, ReturningUpdate, ReturningDelete,
which is fine since the "row-ness" of these constructs
doesn't happen until returning() is called in any case.

a Tuple based model is then integrated so that these
objects can carry along information about their return
types.  Overloads at the .execute() level carry through
the Tuple from the invoked object to the result.

To suit the issue of AliasedClass generating attributes
that are dynamic, experimented with a custom subclass
AsAliased, but then just settled on having aliased()
lie to the type checker and return `Type[_O]`, essentially.
will need some type-related accessors for with_polymorphic()
also.

Additionally, identified an issue in Update when used
"mysql style" against a join(), it basically doesn't work
if asked to UPDATE two tables on the same column name.
added an error message to the specific condition where
it happens with a very non-specific error message that we
hit a thing we can't do right now, suggest multi-table
update as a possible cause.

Change-Id: I5eff7eefe1d6166ee74160b2785c5e6a81fa8b95

4 files changed, 101 insertions, 1 deletions

diff --git a/test/sql/test_metadata.py b/test/sql/test_metadata.py
index b175f9663..1a070fcf5 100644
--- a/test/sql/test_metadata.py
+++ b/test/sql/test_metadata.py
@@ -22,6 +22,7 @@ from sqlalchemy import Integer
 from sqlalchemy import MetaData
 from sqlalchemy import PrimaryKeyConstraint
 from sqlalchemy import schema
+from sqlalchemy import select
 from sqlalchemy import Sequence
 from sqlalchemy import String
 from sqlalchemy import Table
@@ -1337,6 +1338,20 @@ class ToMetaDataTest(fixtures.TestBase, AssertsCompiledSQL, ComparesTables):
 
         self._assert_fk(t2, None, "p.t1.x", referred_schema_fn=ref_fn)
 
+    def test_fk_get_referent_is_always_a_column(self):
+        """test the annotation on ForeignKey.get_referent() in that it does
+        in fact return Column even if given a labeled expr in a subquery"""
+
+        m = MetaData()
+        a = Table("a", m, Column("id", Integer, primary_key=True))
+        b = Table("b", m, Column("aid", Integer, ForeignKey("a.id")))
+
+        stmt = select(a.c.id.label("somelabel")).subquery()
+
+        referent = list(b.c.aid.foreign_keys)[0].get_referent(stmt)
+        is_(referent, stmt.c.somelabel)
+        assert isinstance(referent, Column)
+
     def test_copy_info(self):
         m = MetaData()
         fk = ForeignKey("t2.id")
diff --git a/test/sql/test_resultset.py b/test/sql/test_resultset.py
index ff70fc184..cb9f93018 100644
--- a/test/sql/test_resultset.py
+++ b/test/sql/test_resultset.py
@@ -156,6 +156,50 @@ class CursorResultTest(fixtures.TablesTest):
             rows.append(row)
         eq_(len(rows), 3)
 
+    def test_scalars(self, connection):
+        users = self.tables.users
+
+        connection.execute(
+            users.insert(),
+            [
+                {"user_id": 7, "user_name": "jack"},
+                {"user_id": 8, "user_name": "ed"},
+                {"user_id": 9, "user_name": "fred"},
+            ],
+        )
+        r = connection.scalars(users.select().order_by(users.c.user_id))
+        eq_(r.all(), [7, 8, 9])
+
+    def test_result_tuples(self, connection):
+        users = self.tables.users
+
+        connection.execute(
+            users.insert(),
+            [
+                {"user_id": 7, "user_name": "jack"},
+                {"user_id": 8, "user_name": "ed"},
+                {"user_id": 9, "user_name": "fred"},
+            ],
+        )
+        r = connection.execute(
+            users.select().order_by(users.c.user_id)
+        ).tuples()
+        eq_(r.all(), [(7, "jack"), (8, "ed"), (9, "fred")])
+
+    def test_row_tuple(self, connection):
+        users = self.tables.users
+
+        connection.execute(
+            users.insert(),
+            [
+                {"user_id": 7, "user_name": "jack"},
+                {"user_id": 8, "user_name": "ed"},
+                {"user_id": 9, "user_name": "fred"},
+            ],
+        )
+        r = connection.execute(users.select().order_by(users.c.user_id))
+        eq_([row.t for row in r], [(7, "jack"), (8, "ed"), (9, "fred")])
+
     def test_row_next(self, connection):
         users = self.tables.users
 
diff --git a/test/sql/test_select.py b/test/sql/test_select.py
index be64e205e..d91e50e63 100644
--- a/test/sql/test_select.py
+++ b/test/sql/test_select.py
@@ -61,7 +61,7 @@ class SelectTest(fixtures.TestBase, AssertsCompiledSQL):
     def test_old_bracket_style_fail(self):
         with expect_raises_message(
             exc.ArgumentError,
-            r"Column expression or FROM clause expected, "
+            r"Column expression, FROM clause, or other columns clause .*"
             r".*Did you mean to say",
         ):
             select([table1.c.myid])
diff --git a/test/sql/test_update.py b/test/sql/test_update.py
index 619cbd863..e93900bbd 100644
--- a/test/sql/test_update.py
+++ b/test/sql/test_update.py
@@ -25,6 +25,7 @@ from sqlalchemy.testing import assert_raises
 from sqlalchemy.testing import assert_raises_message
 from sqlalchemy.testing import AssertsCompiledSQL
 from sqlalchemy.testing import eq_
+from sqlalchemy.testing import expect_raises_message
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing import mock
 from sqlalchemy.testing.schema import Column
@@ -1016,6 +1017,46 @@ class UpdateFromCompileTest(
             dialect="mysql",
         )
 
+    def test_update_from_join_unsupported_cases(self):
+        """
+        found_during_typing
+
+        It's unclear how to cleanly guard against this case without producing
+        false positives, particularly due to the support for UPDATE
+        of a CTE.  I'm also not sure of the nature of the failure and why
+        it happens this way.
+
+        """
+        users, addresses = self.tables.users, self.tables.addresses
+
+        j = users.join(addresses)
+
+        with expect_raises_message(
+            exc.CompileError,
+            r"Encountered unsupported case when compiling an INSERT or UPDATE "
+            r"statement.  If this is a multi-table "
+            r"UPDATE statement, please provide string-named arguments to the "
+            r"values\(\) method with distinct names; support for multi-table "
+            r"UPDATE statements that "
+            r"target multiple tables for UPDATE is very limited",
+        ):
+            update(j).where(addresses.c.email_address == "e1").values(
+                {users.c.id: 10, addresses.c.email_address: "asdf"}
+            ).compile(dialect=mysql.dialect())
+
+        with expect_raises_message(
+            exc.CompileError,
+            r"Encountered unsupported case when compiling an INSERT or UPDATE "
+            r"statement.  If this is a multi-table "
+            r"UPDATE statement, please provide string-named arguments to the "
+            r"values\(\) method with distinct names; support for multi-table "
+            r"UPDATE statements that "
+            r"target multiple tables for UPDATE is very limited",
+        ):
+            update(j).where(addresses.c.email_address == "e1").compile(
+                dialect=mysql.dialect()
+            )
+
     def test_update_from_join_mysql_whereclause(self):
         users, addresses = self.tables.users, self.tables.addresses