('select * from t1', {}, None),
('DROP TABLE t1', {}, None)]
- # or engine.dialect.preexecute_pk_sequences:
- if not testing.against('oracle+zxjdbc'):
- cursor = [
- ('CREATE TABLE t1', {}, ()),
- ('INSERT INTO t1 (c1, c2)', {
- 'c2': 'some data', 'c1': 5},
- (5, 'some data')),
- ('SELECT lower', {'lower_2': 'Foo'},
- ('Foo', )),
- ('INSERT INTO t1 (c1, c2)',
- {'c2': 'foo', 'c1': 6},
- (6, 'foo')),
- ('select * from t1', {}, ()),
- ('DROP TABLE t1', {}, ()),
- ]
- else:
- insert2_params = 6, 'Foo'
- if testing.against('oracle+zxjdbc'):
- insert2_params += (ReturningParam(12), )
- cursor = [('CREATE TABLE t1', {}, ()),
- ('INSERT INTO t1 (c1, c2)',
- {'c2': 'some data', 'c1': 5}, (5, 'some data')),
- ('INSERT INTO t1 (c1, c2)',
- {'c1': 6, 'lower_2': 'Foo'}, insert2_params),
- ('select * from t1', {}, ()),
- ('DROP TABLE t1', {}, ())]
- # bind param name 'lower_2' might
- # be incorrect
+ cursor = [
+ ('CREATE TABLE t1', {}, ()),
+ ('INSERT INTO t1 (c1, c2)', {
+ 'c2': 'some data', 'c1': 5},
+ (5, 'some data')),
+ ('SELECT lower', {'lower_1': 'Foo'},
+ ('Foo', )),
+ ('INSERT INTO t1 (c1, c2)',
+ {'c2': 'foo', 'c1': 6},
+ (6, 'foo')),
+ ('select * from t1', {}, ()),
+ ('DROP TABLE t1', {}, ()),
+ ]
self._assert_stmts(compiled, stmts)
self._assert_stmts(cursor, cursor_stmts)
('INSERT INTO t1 (c1, c2)', {'c1': 6}, None),
('select * from t1', {}, None),
('DROP TABLE t1', {}, None)]
- # or engine.dialect.pr eexecute_pk_sequence s:
- # original comment above moved here for pep8 fix
- if not testing.against('oracle+zxjdbc'):
- cursor = [
- ('CREATE TABLE t1', {}, ()),
- ('INSERT INTO t1 (c1, c2)', {
- 'c2': 'some data', 'c1': 5}, (5, 'some data')),
- ('SELECT lower', {'lower_2': 'Foo'},
- ('Foo', )),
- ('INSERT INTO t1 (c1, c2)', {'c2': 'foo', 'c1': 6},
- (6, 'foo')),
- ('select * from t1', {}, ()),
- ('DROP TABLE t1', {}, ()),
- ]
- else:
- insert2_params = 6, 'Foo'
- if testing.against('oracle+zxjdbc'):
- insert2_params += (ReturningParam(12), )
- cursor = [('CREATE TABLE t1', {}, ()),
- ('INSERT INTO t1 (c1, c2)', {
- 'c2': 'some data', 'c1': 5}, (5, 'some data')),
- ('INSERT INTO t1 (c1, c2)',
- {'c1': 6, 'lower_2': 'Foo'}, insert2_params),
- ('select * from t1', {}, ()),
- ('DROP TABLE t1', {}, ())]
+
+ cursor = [
+ ('CREATE TABLE t1', {}, ()),
+ ('INSERT INTO t1 (c1, c2)', {
+ 'c2': 'some data', 'c1': 5}, (5, 'some data')),
+ ('SELECT lower', {'lower_1': 'Foo'},
+ ('Foo', )),
+ ('INSERT INTO t1 (c1, c2)', {'c2': 'foo', 'c1': 6},
+ (6, 'foo')),
+ ('select * from t1', {}, ()),
+ ('DROP TABLE t1', {}, ()),
+ ]
+
assert_stmts(compiled, stmts)
assert_stmts(cursor, cursor_stmts)
from sqlalchemy.testing import engines
from sqlalchemy import (
MetaData, Integer, String, ForeignKey, Boolean, exc, Sequence, func,
- literal, Unicode, cast)
+ literal, Unicode, cast, DateTime)
from sqlalchemy.types import TypeDecorator, TypeEngine
from sqlalchemy.testing.schema import Table, Column
from sqlalchemy.dialects import sqlite
default=sa.select([func.max(t2.c.nextid)]).as_scalar()),
Column('data', String(30)))
+ Table(
+ 'date_table', metadata,
+ Column(
+ 'date_id',
+ DateTime, default=text("current_timestamp"), primary_key=True)
+ )
+
@testing.requires.returning
def test_with_implicit_returning(self):
self._test(True)
self._test(False)
def _test(self, returning):
- t2, t1 = self.tables.t2, self.tables.t1
+ t2, t1, date_table = (
+ self.tables.t2, self.tables.t1, self.tables.date_table
+ )
if not returning and not testing.db.dialect.implicit_returning:
engine = testing.db
else:
engine = engines.testing_engine(
options={'implicit_returning': returning})
- engine.execute(t2.insert(), nextid=1)
- r = engine.execute(t1.insert(), data='hi')
- eq_([1], r.inserted_primary_key)
+ with engine.begin() as conn:
+ conn.execute(t2.insert(), nextid=1)
+ r = conn.execute(t1.insert(), data='hi')
+ eq_([1], r.inserted_primary_key)
+
+ conn.execute(t2.insert(), nextid=2)
+ r = conn.execute(t1.insert(), data='there')
+ eq_([2], r.inserted_primary_key)
- engine.execute(t2.insert(), nextid=2)
- r = engine.execute(t1.insert(), data='there')
- eq_([2], r.inserted_primary_key)
+ r = conn.execute(date_table.insert())
+ assert isinstance(r.inserted_primary_key[0], datetime.datetime)
class PKIncrementTest(fixtures.TablesTest):
def test_literal_default_no_label(self):
self._run_test(default=literal("INT_1", type_=self.MyInteger))
+ def test_literal_column_default_no_label(self):
+ self._run_test(default=literal_column("1", type_=self.MyInteger))
+
def test_sequence(self):
self._run_test(Sequence('foo_seq'))
+ def test_text_clause_default_no_type(self):
+ self._run_test(default=text('1'))
+
def test_server_default(self):
self._run_test(server_default='1',)