small tweaks to make insert() behavior more consistent, mostly tests, [ticket:2461]

diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py
index 1c74dab43fc0ded893ef0a3048956aa3350cbf0b..4bb2427c7da055db6becae083ae4f9b3906a8185 100644 (file)
--- a/lib/sqlalchemy/engine/default.py
+++ b/lib/sqlalchemy/engine/default.py
@@ -671,13 +671,13 @@ class DefaultExecutionContext(interfaces.ExecutionContext):
 
     def post_insert(self):
         if not self._is_implicit_returning and \
+            not self.compiled.inline and \
             self.dialect.postfetch_lastrowid and \
             (not self.inserted_primary_key or \
                         None in self.inserted_primary_key):
 
             table = self.compiled.statement.table
             lastrowid = self.get_lastrowid()
-
             autoinc_col = table._autoincrement_column
             if autoinc_col is not None:
                 # apply type post processors to the lastrowid

diff --git a/lib/sqlalchemy/sql/expression.py b/lib/sqlalchemy/sql/expression.py
index cbc3b47ad22479815bd590855d66bd64a7e5a6a1..b7b965ea9c4046b8d757056a7cbe9ddfebc25699 100644 (file)
--- a/lib/sqlalchemy/sql/expression.py
+++ b/lib/sqlalchemy/sql/expression.py
@@ -4455,6 +4455,13 @@ class TableClause(Immutable, FromClause):
 
     named_with_column = True
 
+    implicit_returning = False
+    """:class:`.TableClause` doesn't support having a primary key or column
+    -level defaults, so implicit returning doesn't apply."""
+
+    _autoincrement_column = None
+    """No PK or default support so no autoincrement column."""
+
     def __init__(self, name, *columns):
         super(TableClause, self).__init__()
         self.name = self.fullname = name

diff --git a/test/lib/fixtures.py b/test/lib/fixtures.py
index 15a3d6b0322779e93ccd7b1a3c92909f592b7eed..00d35df75213edb25fa1bcd25547b8ac093f2ad7 100644 (file)
--- a/test/lib/fixtures.py
+++ b/test/lib/fixtures.py
@@ -108,6 +108,9 @@ class TablesTest(TestBase):
             if self.run_create_tables == 'each':
                 self.metadata.create_all(self.bind)
             self.tables.update(self.metadata.tables)
+        elif self.run_create_tables == 'each':
+            drop_all_tables(self.metadata, self.bind)
+            self.metadata.create_all(self.bind)
 
     def _setup_each_inserts(self):
         if self.run_inserts == 'each':

diff --git a/test/sql/test_query.py b/test/sql/test_query.py
index 16a2f254d77e6ab13d76f6baa8f10eba2f2996a1..06854c4fcabbe80a01c28236a827f3dc47efb953 100644 (file)
--- a/test/sql/test_query.py
+++ b/test/sql/test_query.py
@@ -1214,6 +1214,155 @@ class QueryTest(fixtures.TestBase):
         r = s.execute().fetchall()
         assert len(r) == 1
 
+class TableInsertTest(fixtures.TablesTest):
+    """test for consistent insert behavior across dialects
+    regarding the inline=True flag, lower-case 't' tables.
+
+    """
+    run_create_tables = 'each'
+
+    @classmethod
+    def define_tables(cls, metadata):
+        Table('foo', metadata,
+                Column('id', Integer, Sequence('t_id_seq'), primary_key=True),
+                Column('data', String(50)),
+                Column('x', Integer)
+            )
+
+    def _fixture(self, types=True):
+        if types:
+            t = sql.table('foo', sql.column('id', Integer),
+                        sql.column('data', String),
+                        sql.column('x', Integer))
+        else:
+            t = sql.table('foo', sql.column('id'),
+                        sql.column('data'),
+                        sql.column('x'))
+        return t
+
+    def _test(self, stmt, row, returning=None, inserted_primary_key=False):
+        r = testing.db.execute(stmt)
+
+        if returning:
+            returned = r.first()
+            eq_(returned, returning)
+        elif inserted_primary_key is not False:
+            eq_(r.inserted_primary_key, inserted_primary_key)
+
+        eq_(testing.db.execute(self.tables.foo.select()).first(), row)
+
+    def _test_multi(self, stmt, rows, data):
+        testing.db.execute(stmt, rows)
+        eq_(
+            testing.db.execute(self.tables.foo.select().
+                            order_by(self.tables.foo.c.id)).fetchall(),
+            data)
+
+    @testing.requires.sequences
+    def test_expicit_sequence(self):
+        t = self._fixture()
+        self._test(
+            t.insert().values(
+                        id=func.next_value(Sequence('t_id_seq')),
+                        data='data', x=5
+                    ),
+            (1, 'data', 5)
+        )
+
+    def test_uppercase(self):
+        t = self.tables.foo
+        self._test(
+            t.insert().values(
+                        id=1,
+                        data='data', x=5
+                    ),
+            (1, 'data', 5),
+            inserted_primary_key=[1]
+        )
+
+    def test_uppercase_inline(self):
+        t = self.tables.foo
+        self._test(
+            t.insert(inline=True).values(
+                        id=1,
+                        data='data', x=5
+                    ),
+            (1, 'data', 5),
+            inserted_primary_key=[1]
+        )
+
+    def test_uppercase_inline_implicit(self):
+        t = self.tables.foo
+        self._test(
+            t.insert(inline=True).values(
+                        data='data', x=5
+                    ),
+            (1, 'data', 5),
+            inserted_primary_key=[None]
+        )
+
+    def test_uppercase_implicit(self):
+        t = self.tables.foo
+        self._test(
+            t.insert().values(data='data', x=5),
+            (1, 'data', 5),
+            inserted_primary_key=[1]
+        )
+
+    def test_direct_params(self):
+        t = self._fixture()
+        self._test(
+            t.insert().values(id=1, data='data', x=5),
+            (1, 'data', 5),
+            inserted_primary_key=[]
+        )
+
+    @testing.requires.returning
+    def test_direct_params_returning(self):
+        t = self._fixture()
+        self._test(
+            t.insert().values(
+                        id=1, data='data', x=5).returning(t.c.id, t.c.x),
+            (1, 'data', 5),
+            returning=(1, 5)
+        )
+
+    @testing.requires.dbapi_lastrowid
+    def test_implicit_pk(self):
+        t = self._fixture()
+        self._test(
+            t.insert().values(
+                        data='data', x=5),
+            (1, 'data', 5),
+            inserted_primary_key=[]
+        )
+
+    @testing.requires.dbapi_lastrowid
+    def test_implicit_pk_multi_rows(self):
+        t = self._fixture()
+        self._test_multi(
+            t.insert(),
+            [
+                {'data':'d1', 'x':5},
+                {'data':'d2', 'x':6},
+                {'data':'d3', 'x':7},
+            ],
+            [
+                (1, 'd1', 5),
+                (2, 'd2', 6),
+                (3, 'd3', 7)
+            ],
+        )
+
+    @testing.requires.dbapi_lastrowid
+    def test_implicit_pk_inline(self):
+        t = self._fixture()
+        self._test(
+            t.insert(inline=True).values(data='data', x=5),
+            (1, 'data', 5),
+            inserted_primary_key=[]
+        )
+
 class PercentSchemaNamesTest(fixtures.TestBase):
     """tests using percent signs, spaces in table and column names.