c = self.proxy("select %s" % column.default.arg)
return c.fetchone()[0]
elif isinstance(column.type, types.Integer) and (column.default is None or (isinstance(column.default, schema.Sequence) and column.default.optional)):
- c = self.proxy("select nextval('%s_%s_seq')" % (column.table.name, column.name))
+ sch = column.table.schema
+ if sch is not None:
+ exc = "select nextval('%s.%s_%s_seq')" % (sch, column.table.name, column.name)
+ else:
+ exc = "select nextval('%s_%s_seq')" % (column.table.name, column.name)
+ c = self.proxy(exc)
return c.fetchone()[0]
else:
return ansisql.ANSIDefaultRunner.get_column_default(self, column)
self.assert_(l == [(1, 'user', 'lala')])
finally:
db.execute("drop table speedy_users", None)
+
+ def testschema(self):
+ if not db.engine.__module__.endswith('postgres'):
+ return
+
+ test_table = Table('my_table', db,
+ Column('id', Integer, primary_key=True),
+ Column('data', String(20), nullable=False),
+ schema='alt_schema'
+ )
+ test_table.create()
+ try:
+ # plain insert
+ test_table.insert().execute(data='test')
+
+ # try with a PassiveDefault
+ test_table.deregister()
+ test_table = Table('my_table', db, autoload=True, redefine=True, schema='alt_schema')
+ test_table.insert().execute(data='test')
+
+ finally:
+ test_table.drop()
+
def testdefaults(self):
x = {'x':50}
def mydefault():