From: Mike Bayer Date: Thu, 23 Aug 2012 22:49:33 +0000 (-0400) Subject: small tweaks to make insert() behavior more consistent, mostly tests, [ticket:2461] X-Git-Tag: rel_0_8_0b1~213 X-Git-Url: http://git.ipfire.org/gitweb.cgi?a=commitdiff_plain;h=17f9bc5735b021201c1800adc2236b3fe67262b2;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git 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 1c74dab43f..4bb2427c7d 100644 --- 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 cbc3b47ad2..b7b965ea9c 100644 --- 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 15a3d6b032..00d35df752 100644 --- 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 16a2f254d7..06854c4fca 100644 --- 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.