- Added new method to the :func:`.insert` construct

:meth:`.Insert.from_select`.  Given a list of columns and
a selectable, renders ``INSERT INTO (table) (columns) SELECT ..``.
While this feature is highlighted as part of 0.9 it is also
backported to 0.8.3. [ticket:722]
- The :func:`.update`, :func:`.insert`, and :func:`.delete` constructs
will now interpret ORM entities as FROM clauses to be operated upon,
in the same way that select() already does.  Also in 0.8.3.

2 files changed, 134 insertions, 2 deletions

diff --git a/test/orm/test_query.py b/test/orm/test_query.py
index 61ce39c04..0973dc357 100644
--- a/test/orm/test_query.py
+++ b/test/orm/test_query.py
@@ -1,7 +1,7 @@
 from sqlalchemy.sql import operators
 from sqlalchemy import MetaData, null, exists, text, union, literal, \
     literal_column, func, between, Unicode, desc, and_, bindparam, \
-    select, distinct, or_, collate
+    select, distinct, or_, collate, insert
 from sqlalchemy import inspect
 from sqlalchemy import exc as sa_exc, util
 from sqlalchemy.sql import compiler, table, column
@@ -265,6 +265,75 @@ class RawSelectTest(QueryTest, AssertsCompiledSQL):
             "JOIN addresses ON users.id = addresses.user_id"
         )
 
+    def test_insert_from_query(self):
+        User = self.classes.User
+        Address = self.classes.Address
+
+        s = Session()
+        q = s.query(User.id, User.name).filter_by(name='ed')
+        self.assert_compile(
+            insert(Address).from_select(('id', 'email_address'), q),
+            "INSERT INTO addresses (id, email_address) "
+            "SELECT users.id AS users_id, users.name AS users_name "
+            "FROM users WHERE users.name = :name_1"
+        )
+
+    def test_insert_from_query_col_attr(self):
+        User = self.classes.User
+        Address = self.classes.Address
+
+        s = Session()
+        q = s.query(User.id, User.name).filter_by(name='ed')
+        self.assert_compile(
+            insert(Address).from_select(
+                            (Address.id, Address.email_address), q),
+            "INSERT INTO addresses (id, email_address) "
+            "SELECT users.id AS users_id, users.name AS users_name "
+            "FROM users WHERE users.name = :name_1"
+        )
+
+    def test_update_from_entity(self):
+        from sqlalchemy.sql import update
+        User = self.classes.User
+        self.assert_compile(
+            update(User),
+            "UPDATE users SET id=:id, name=:name"
+        )
+
+        self.assert_compile(
+            update(User).values(name='ed').where(User.id == 5),
+            "UPDATE users SET name=:name WHERE users.id = :id_1",
+            checkparams={"id_1": 5, "name": "ed"}
+        )
+
+    def test_delete_from_entity(self):
+        from sqlalchemy.sql import delete
+        User = self.classes.User
+        self.assert_compile(
+            delete(User),
+            "DELETE FROM users"
+        )
+
+        self.assert_compile(
+            delete(User).where(User.id == 5),
+            "DELETE FROM users WHERE users.id = :id_1",
+            checkparams={"id_1": 5}
+        )
+
+    def test_insert_from_entity(self):
+        from sqlalchemy.sql import insert
+        User = self.classes.User
+        self.assert_compile(
+            insert(User),
+            "INSERT INTO users (id, name) VALUES (:id, :name)"
+        )
+
+        self.assert_compile(
+            insert(User).values(name="ed"),
+            "INSERT INTO users (name) VALUES (:name)",
+            checkparams={"name": "ed"}
+        )
+
 class GetTest(QueryTest):
     def test_get(self):
         User = self.classes.User
diff --git a/test/sql/test_insert.py b/test/sql/test_insert.py
index cd040538f..e1171532d 100644
--- a/test/sql/test_insert.py
+++ b/test/sql/test_insert.py
@@ -1,7 +1,7 @@
 #! coding:utf-8
 
 from sqlalchemy import Column, Integer, MetaData, String, Table,\
-    bindparam, exc, func, insert
+    bindparam, exc, func, insert, select
 from sqlalchemy.dialects import mysql, postgresql
 from sqlalchemy.engine import default
 from sqlalchemy.testing import AssertsCompiledSQL,\
@@ -120,6 +120,69 @@ class InsertTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
             dialect=default.DefaultDialect()
         )
 
+    def test_insert_from_select_select(self):
+        table1 = self.tables.mytable
+        sel = select([table1.c.myid, table1.c.name]).where(table1.c.name == 'foo')
+        ins = self.tables.myothertable.insert().\
+                    from_select(("otherid", "othername"), sel)
+        self.assert_compile(
+            ins,
+            "INSERT INTO myothertable (otherid, othername) "
+            "SELECT mytable.myid, mytable.name FROM mytable "
+            "WHERE mytable.name = :name_1",
+            checkparams={"name_1": "foo"}
+        )
+
+    def test_insert_mix_select_values_exception(self):
+        table1 = self.tables.mytable
+        sel = select([table1.c.myid, table1.c.name]).where(table1.c.name == 'foo')
+        ins = self.tables.myothertable.insert().\
+                    from_select(("otherid", "othername"), sel)
+        assert_raises_message(
+            exc.InvalidRequestError,
+            "This construct already inserts from a SELECT",
+            ins.values, othername="5"
+        )
+
+    def test_insert_mix_values_select_exception(self):
+        table1 = self.tables.mytable
+        sel = select([table1.c.myid, table1.c.name]).where(table1.c.name == 'foo')
+        ins = self.tables.myothertable.insert().values(othername="5")
+        assert_raises_message(
+            exc.InvalidRequestError,
+            "This construct already inserts value expressions",
+            ins.from_select, ("otherid", "othername"), sel
+        )
+
+    def test_insert_from_select_table(self):
+        table1 = self.tables.mytable
+        ins = self.tables.myothertable.insert().\
+                    from_select(("otherid", "othername"), table1)
+        # note we aren't checking the number of columns right now
+        self.assert_compile(
+            ins,
+            "INSERT INTO myothertable (otherid, othername) "
+            "SELECT mytable.myid, mytable.name, mytable.description "
+            "FROM mytable",
+            checkparams={}
+        )
+
+
+    def test_insert_from_select_col_values(self):
+        table1 = self.tables.mytable
+        table2 = self.tables.myothertable
+        sel = select([table1.c.myid, table1.c.name]).where(table1.c.name == 'foo')
+        ins = table2.insert().\
+                    from_select((table2.c.otherid, table2.c.othername), sel)
+        self.assert_compile(
+            ins,
+            "INSERT INTO myothertable (otherid, othername) "
+            "SELECT mytable.myid, mytable.name FROM mytable "
+            "WHERE mytable.name = :name_1",
+            checkparams={"name_1": "foo"}
+        )
+
+
 class EmptyTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
     __dialect__ = 'default'