- Added new generic function "next_value()", accepts

a Sequence object as its argument and renders the
appropriate "next value" generation string on the
target platform, if supported.  Also provides
".next_value()" method on Sequence itself.
[ticket:2085]
- added tests for all the conditions described
in [ticket:2085]
- postgresql dialect will exec/compile a Sequence
that has "optional=True".  the optional flag is now only
checked specifically in the context of a Table primary key
evaulation.
- func.next_value() or other SQL expression can
be embedded directly into an insert() construct,
and if implicit or explicit "returning" is used
in conjunction with a primary key column,
the newly generated value will be present in
result.inserted_primary_key. [ticket:2084]

1 files changed, 184 insertions, 26 deletions

diff --git a/test/sql/test_defaults.py b/test/sql/test_defaults.py
index eeb4ebbcb..79243ad4f 100644
--- a/test/sql/test_defaults.py
+++ b/test/sql/test_defaults.py
@@ -10,7 +10,7 @@ from sqlalchemy.types import TypeDecorator
 from test.lib.schema import Table, Column
 from test.lib.testing import eq_
 from test.sql import _base
-
+from sqlalchemy.dialects import sqlite
 
 class DefaultTest(testing.TestBase):
 
@@ -303,6 +303,31 @@ class DefaultTest(testing.TestBase):
              (53, 'imthedefault', f, ts, ts, ctexec, True, False,
               12, today, 'py')])
 
+    def test_no_embed_in_sql(self):
+        """Using a DefaultGenerator, Sequence, DefaultClause
+        in the columns, where clause of a select, or in the values 
+        clause of insert, update, raises an informative error"""
+        for const in (
+            sa.Sequence('y'),
+            sa.ColumnDefault('y'),
+            sa.DefaultClause('y')
+        ):
+            assert_raises_message(
+                sa.exc.ArgumentError,
+                "SQL expression object or string expected.",
+                t.select, [const]
+            )
+            assert_raises_message(
+                sa.exc.InvalidRequestError,
+                "cannot be used directly as a column expression.",
+                str, t.insert().values(col4=const)
+            )
+            assert_raises_message(
+                sa.exc.InvalidRequestError,
+                "cannot be used directly as a column expression.",
+                str, t.update().values(col4=const)
+            )
+
     def test_missing_many_param(self):
         assert_raises_message(exc.StatementError, 
             "A value is required for bind parameter 'col7', in parameter group 1",
@@ -565,6 +590,132 @@ class SequenceDDLTest(testing.TestBase, testing.AssertsCompiledSQL):
             "DROP SEQUENCE foo_seq",
         )
 
+class SequenceExecTest(testing.TestBase):
+    __requires__ = ('sequences',)
+
+    @classmethod
+    def setup_class(cls):
+        cls.seq= Sequence("my_sequence")
+        cls.seq.create(testing.db)
+
+    @classmethod
+    def teardown_class(cls):
+        cls.seq.drop(testing.db)
+
+    def _assert_seq_result(self, ret):
+        """asserts return of next_value is an int"""
+        assert isinstance(ret, (int, long))
+        assert ret > 0
+
+    def test_implicit_connectionless(self):
+        s = Sequence("my_sequence", metadata=MetaData(testing.db))
+        self._assert_seq_result(s.execute())
+
+    def test_explicit(self):
+        s = Sequence("my_sequence")
+        self._assert_seq_result(s.execute(testing.db))
+
+    def test_explicit_optional(self):
+        """test dialect executes a Sequence, returns nextval, whether 
+        or not "optional" is set """
+
+        s = Sequence("my_sequence", optional=True)
+        self._assert_seq_result(s.execute(testing.db))
+
+    def test_func_implicit_connectionless_execute(self):
+        """test func.next_value().execute()/.scalar() works
+        with connectionless execution. """
+
+        s = Sequence("my_sequence", metadata=MetaData(testing.db))
+        self._assert_seq_result(s.next_value().execute().scalar())
+
+    def test_func_explicit(self):
+        s = Sequence("my_sequence")
+        self._assert_seq_result(testing.db.scalar(s.next_value()))
+
+    def test_func_implicit_connectionless_scalar(self):
+        """test func.next_value().execute()/.scalar() works. """
+
+        s = Sequence("my_sequence", metadata=MetaData(testing.db))
+        self._assert_seq_result(s.next_value().scalar())
+
+    def test_func_embedded_select(self):
+        """test can use next_value() in select column expr"""
+
+        s = Sequence("my_sequence")
+        self._assert_seq_result(
+            testing.db.scalar(select([s.next_value()]))
+        )
+
+    @testing.fails_on('oracle', "ORA-02287: sequence number not allowed here")
+    @testing.provide_metadata
+    def test_func_embedded_whereclause(self):
+        """test can use next_value() in whereclause"""
+
+        t1 = Table('t', metadata,
+            Column('x', Integer)
+        )
+        t1.create(testing.db)
+        testing.db.execute(t1.insert(), [{'x':1}, {'x':300}, {'x':301}])
+        s = Sequence("my_sequence")
+        eq_(
+            testing.db.execute(
+                t1.select().where(t1.c.x > s.next_value())
+            ).fetchall(),
+            [(300, ), (301, )]
+        )
+
+    @testing.provide_metadata
+    def test_func_embedded_valuesbase(self):
+        """test can use next_value() in values() of _ValuesBase"""
+
+        t1 = Table('t', metadata,
+            Column('x', Integer)
+        )
+        t1.create(testing.db)
+        s = Sequence("my_sequence")
+        testing.db.execute(
+            t1.insert().values(x=s.next_value())
+        )
+        self._assert_seq_result(
+            testing.db.scalar(t1.select())
+        )
+
+    @testing.provide_metadata
+    def test_inserted_pk_no_returning(self):
+        """test inserted_primary_key contains [None] when 
+        pk_col=next_value(), implicit returning is not used."""
+
+        e = engines.testing_engine(options={'implicit_returning':False})
+        s = Sequence("my_sequence")
+        metadata.bind = e
+        t1 = Table('t', metadata,
+            Column('x', Integer, primary_key=True)
+        )
+        t1.create()
+        r = e.execute(
+            t1.insert().values(x=s.next_value())
+        )
+        eq_(r.inserted_primary_key, [None])
+
+    @testing.requires.returning
+    @testing.provide_metadata
+    def test_inserted_pk_implicit_returning(self):
+        """test inserted_primary_key contains the result when 
+        pk_col=next_value(), when implicit returning is used."""
+
+        e = engines.testing_engine(options={'implicit_returning':True})
+        s = Sequence("my_sequence")
+        metadata.bind = e
+        t1 = Table('t', metadata,
+            Column('x', Integer, primary_key=True)
+        )
+        t1.create()
+        r = e.execute(
+            t1.insert().values(x=s.next_value())
+        )
+        self._assert_seq_result(r.inserted_primary_key[0])
+
 class SequenceTest(testing.TestBase, testing.AssertsCompiledSQL):
     __requires__ = ('sequences',)
 
@@ -586,29 +737,36 @@ class SequenceTest(testing.TestBase, testing.AssertsCompiledSQL):
             finally:
                 seq.drop(testing.db)
 
-    @testing.fails_on('maxdb', 'FIXME: unknown')
-    # maxdb db-api seems to double-execute NEXTVAL 
-    # internally somewhere,
-    # throwing off the numbers for these tests...
-    @testing.provide_metadata
-    def test_implicit_sequence_exec(self):
-        s = Sequence("my_sequence", metadata=metadata)
-        metadata.create_all()
-        x = s.execute()
-        eq_(x, 1)
 
     def _has_sequence(self, name):
         return testing.db.dialect.has_sequence(testing.db, name)
 
-    @testing.fails_on('maxdb', 'FIXME: unknown')
-    def teststandalone_explicit(self):
-        s = Sequence("my_sequence")
-        s.create(bind=testing.db)
-        try:
-            x = s.execute(testing.db)
-            eq_(x, 1)
-        finally:
-            s.drop(testing.db)
+    def test_nextval_render(self):
+        """test dialect renders the "nextval" construct, 
+        whether or not "optional" is set """
+
+        for s in (
+                Sequence("my_seq"), 
+                Sequence("my_seq", optional=True)):
+            assert str(s.next_value().
+                    compile(dialect=testing.db.dialect)) in (
+                "nextval('my_seq')",
+                "gen_id(my_seq, 1)",
+                "my_seq.nextval",
+            )
+
+    def test_nextval_unsupported(self):
+        """test next_value() used on non-sequence platform 
+        raises NotImplementedError."""
+
+        s = Sequence("my_seq")
+        d = sqlite.dialect()
+        assert_raises_message(
+            NotImplementedError,
+            "Dialect 'sqlite' does not support sequence increments.",
+            s.next_value().compile,
+            dialect=d
+        )
 
     def test_checkfirst_sequence(self):
         s = Sequence("my_sequence")
@@ -733,10 +891,10 @@ class TableBoundSequenceTest(testing.TestBase):
 
 class SpecialTypePKTest(testing.TestBase):
     """test process_result_value in conjunction with primary key columns.
-    
+
     Also tests that "autoincrement" checks are against column.type._type_affinity,
     rather than the class of "type" itself.
-    
+
     """
 
     @classmethod
@@ -818,12 +976,12 @@ class ServerDefaultsOnPKTest(testing.TestBase):
     @testing.provide_metadata
     def test_string_default_none_on_insert(self):
         """Test that without implicit returning, we return None for 
-        a string server default.  
-        
+        a string server default.
+
         That is, we don't want to attempt to pre-execute "server_default"
         generically - the user should use a Python side-default for a case
         like this.   Testing that all backends do the same thing here.
-        
+
         """
         t = Table('x', metadata, 
                 Column('y', String(10), server_default='key_one', primary_key=True),
@@ -946,7 +1104,7 @@ class UnicodeDefaultsTest(testing.TestBase):
         # end Py2K
         c = Column(Unicode(32), default=default)
 
-    
+
     def test_nonunicode_default(self):
         # Py3K
         #default = b'foo'