or explicit sequence-stated schema, into account.
[ticket:1576]
This is postgresql.DOUBLE_PRECISION in 0.6.
[ticket:1085]
+ - Corrected the "has_sequence" query to take current schema,
+ or explicit sequence-stated schema, into account.
+ [ticket:1576]
+
- mssql
- Changed the name of TrustedConnection to
Trusted_Connection when constructing pyodbc connect
c = connection.execute(tblqry, [self.denormalize_name(table_name)])
return c.first() is not None
- def has_sequence(self, connection, sequence_name):
+ def has_sequence(self, connection, sequence_name, schema=None):
"""Return ``True`` if the given sequence (generator) exists."""
genqry = """
)
return bool(cursor.first())
- def has_sequence(self, connection, sequence_name):
- cursor = connection.execute(
- sql.text("SELECT relname FROM pg_class WHERE relkind = 'S' AND "
- "relnamespace IN ( SELECT oid FROM pg_namespace WHERE nspname NOT LIKE 'pg_%' "
- "AND nspname != 'information_schema' AND relname = :seqname)",
- bindparams=[sql.bindparam('seqname', unicode(sequence_name), type_=sqltypes.Unicode)]
- ))
+ def has_sequence(self, connection, sequence_name, schema=None):
+ if schema is None:
+ cursor = connection.execute(
+ sql.text("SELECT relname FROM pg_class c join pg_namespace n on "
+ "n.oid=c.relnamespace where relkind='S' and n.nspname=current_schema() and lower(relname)=:name",
+ bindparams=[sql.bindparam('name', unicode(sequence_name.lower()), type_=sqltypes.Unicode)]
+ )
+ )
+ else:
+ cursor = connection.execute(
+ sql.text("SELECT relname FROM pg_class c join pg_namespace n on "
+ "n.oid=c.relnamespace where relkind='S' and n.nspname=:schema and lower(relname)=:name",
+ bindparams=[sql.bindparam('name', unicode(sequence_name.lower()), type_=sqltypes.Unicode),
+ sql.bindparam('schema', unicode(schema), type_=sqltypes.Unicode)]
+ )
+ )
+
return bool(cursor.first())
def table_names(self, connection, schema):
if ((not self.dialect.sequences_optional or
not sequence.optional) and
(not self.checkfirst or
- not self.dialect.has_sequence(self.connection, sequence.name))):
+ not self.dialect.has_sequence(self.connection, sequence.name, schema=sequence.schema))):
self.connection.execute(schema.CreateSequence(sequence))
def visit_index(self, index):
if ((not self.dialect.sequences_optional or
not sequence.optional) and
(not self.checkfirst or
- self.dialect.has_sequence(self.connection, sequence.name))):
+ self.dialect.has_sequence(self.connection, sequence.name, schema=sequence.schema))):
self.connection.execute(schema.DropSequence(sequence))
class HasSequenceTest(TestBase):
- @classmethod
- def setup_class(cls):
- global metadata, users
+
+ @testing.requires.sequences
+ def test_has_sequence(self):
metadata = MetaData()
users = Table('users', metadata,
Column('user_id', sa.Integer, sa.Sequence('user_id_seq'), primary_key=True),
Column('user_name', sa.String(40)),
)
+ metadata.create_all(bind=testing.db)
+ try:
+ eq_(testing.db.dialect.has_sequence(testing.db, 'user_id_seq'), True)
+ finally:
+ metadata.drop_all(bind=testing.db)
+ eq_(testing.db.dialect.has_sequence(testing.db, 'user_id_seq'), False)
+ @testing.requires.schemas
@testing.requires.sequences
- def test_hassequence(self):
- metadata.create_all(bind=testing.db)
+ def test_has_sequence_schema(self):
+ test_schema = get_schema()
+ s1 = sa.Sequence('user_id_seq', schema=test_schema)
+ s2 = sa.Sequence('user_id_seq')
+ testing.db.execute(schema.CreateSequence(s1))
+ testing.db.execute(schema.CreateSequence(s2))
+ eq_(testing.db.dialect.has_sequence(testing.db, 'user_id_seq', schema=test_schema), True)
eq_(testing.db.dialect.has_sequence(testing.db, 'user_id_seq'), True)
- metadata.drop_all(bind=testing.db)
+ testing.db.execute(schema.DropSequence(s1))
+ eq_(testing.db.dialect.has_sequence(testing.db, 'user_id_seq', schema=test_schema), False)
+ eq_(testing.db.dialect.has_sequence(testing.db, 'user_id_seq'), True)
+ testing.db.execute(schema.DropSequence(s2))
+ eq_(testing.db.dialect.has_sequence(testing.db, 'user_id_seq', schema=test_schema), False)
eq_(testing.db.dialect.has_sequence(testing.db, 'user_id_seq'), False)
-
+
# Tests related to engine.reflection
def get_schema():