from sqlalchemy import exc
import sqlalchemy as sa
from sqlalchemy.dialects.postgresql import base as postgresql
-from sqlalchemy.dialects.postgresql import ARRAY, INTERVAL
+from sqlalchemy.dialects.postgresql import ARRAY, INTERVAL, TSRANGE
+from sqlalchemy.dialects.postgresql import ExcludeConstraint
import re
self.assert_('uc_a' not in indexes)
self.assert_('uc_a' in constraints)
+ @testing.requires.btree_gist
+ @testing.provide_metadata
+ def test_reflection_with_exclude_constraint(self):
+ m = self.metadata
+ Table(
+ 't', m,
+ Column('id', Integer, primary_key=True),
+ Column('period', TSRANGE),
+ ExcludeConstraint(('period', '&&'), name='quarters_period_excl')
+ )
+
+ m.create_all()
+
+ insp = inspect(testing.db)
+
+ # PostgreSQL will create an implicit index for an exclude constraint.
+ # we don't reflect the EXCLUDE yet.
+ eq_(
+ insp.get_indexes('t'),
+ [{'unique': False, 'name': 'quarters_period_excl',
+ 'duplicates_constraint': 'quarters_period_excl',
+ 'dialect_options': {'postgresql_using': 'gist'},
+ 'column_names': ['period']}]
+ )
+
+ # reflection corrects for the dupe
+ reflected = Table('t', MetaData(testing.db), autoload=True)
+
+ eq_(set(reflected.indexes), set())
+
@testing.provide_metadata
def test_reflect_unique_index(self):
insp = inspect(testing.db)
})
+
class CustomTypeReflectionTest(fixtures.TestBase):
class CustomType(object):
def duplicate_key_raises_integrity_error(self):
return fails_on("postgresql+pg8000")
- @property
- def hstore(self):
- def check_hstore(config):
+ def _has_pg_extension(self, name):
+ def check(config):
if not against(config, "postgresql"):
return False
- try:
- config.db.execute("SELECT 'a=>1,a=>2'::hstore;")
- return True
- except Exception:
- return False
+ count = config.db.scalar(
+ "SELECT count(*) FROM pg_extension "
+ "WHERE extname='%s'" % name)
+ return bool(count)
+ return only_if(check, "needs %s extension" % name)
- return only_if(check_hstore)
+ @property
+ def hstore(self):
+ return self._has_pg_extension("hstore")
+
+ @property
+ def btree_gist(self):
+ return self._has_pg_extension("btree_gist")
@property
def range_types(self):