Calls to reflect a table did not create any UniqueConstraint objects.
The reflection core made no calls to get_unique_constraints and as
a result, the sqlite dialect would never reflect any unique constraints.
MySQL transparently converts unique constraints into unique indexes, but
SQLAlchemy would reflect those as an Index object and as a
UniqueConstraint. The reflection core will now deduplicate the unique
constraints.
PostgreSQL would reflect unique constraints as an Index object and as
a UniqueConstraint object. The reflection core will now deduplicate
the unique indexes.
return [
{
'name': key['name'],
- 'column_names': [col[0] for col in key['columns']]
+ 'column_names': [col[0] for col in key['columns']],
+ 'duplicates_index': key['name'],
}
for key in parsed_state.keys
if key['type'] == 'UNIQUE'
SELECT
i.relname as relname,
ix.indisunique, ix.indexprs, ix.indpred,
- a.attname, a.attnum, ix.indkey%s
+ a.attname, a.attnum, c.conrelid, ix.indkey%s
FROM
pg_class t
join pg_index ix on t.oid = ix.indrelid
- join pg_class i on i.oid=ix.indexrelid
+ join pg_class i on i.oid = ix.indexrelid
left outer join
pg_attribute a
- on t.oid=a.attrelid and %s
+ on t.oid = a.attrelid and %s
+ left outer join
+ pg_constraint c
+ on (ix.indrelid = c.conrelid and
+ ix.indexrelid = c.conindid and
+ c.contype in ('p', 'u', 'x'))
WHERE
t.relkind IN ('r', 'v', 'f', 'm')
and t.oid = :table_oid
sv_idx_name = None
for row in c.fetchall():
- idx_name, unique, expr, prd, col, col_num, idx_key = row
+ idx_name, unique, expr, prd, col, col_num, conrelid, idx_key = row
if expr:
if idx_name != sv_idx_name:
index['cols'][col_num] = col
index['key'] = [int(k.strip()) for k in idx_key.split()]
index['unique'] = unique
+ index['duplicates_constraint'] = (None if conrelid is None
+ else idx_name)
return [
{'name': name,
'unique': idx['unique'],
- 'column_names': [idx['cols'][i] for i in idx['key']]}
+ 'column_names': [idx['cols'][i] for i in idx['key']],
+ 'duplicates_constraint': idx['duplicates_constraint']}
for name, idx in indexes.items()
]
columns = index_d['column_names']
unique = index_d['unique']
flavor = index_d.get('type', 'index')
+ duplicates = index_d.get('duplicates_constraint')
if include_columns and \
not set(columns).issubset(include_columns):
util.warn(
"Omitting %s key for (%s), key covers omitted columns." %
(flavor, ', '.join(columns)))
continue
+ if duplicates:
+ continue
# look for columns by orig name in cols_by_orig_name,
# but support columns that are in-Python only as fallback
idx_cols = []
idx_cols.append(idx_col)
sa_schema.Index(name, *idx_cols, **dict(unique=unique))
+
+ # Unique Constraints
+ constraints = self.get_unique_constraints(table_name, schema)
+ for const_d in constraints:
+ conname = const_d['name']
+ columns = const_d['column_names']
+ duplicates = const_d.get('duplicates_index')
+ if include_columns and \
+ not set(columns).issubset(include_columns):
+ util.warn(
+ "Omitting unique constraint key for (%s), "
+ "key covers omitted columns." %
+ ', '.join(columns))
+ continue
+ if duplicates:
+ continue
+ # look for columns by orig name in cols_by_orig_name,
+ # but support columns that are in-Python only as fallback
+ constrained_cols = []
+ for c in columns:
+ try:
+ constrained_col = cols_by_orig_name[c] \
+ if c in cols_by_orig_name else table.c[c]
+ except KeyError:
+ util.warn(
+ "unique constraint key '%s' was not located in "
+ "columns for table '%s'" % (c, table_name))
+ else:
+ constrained_cols.append(constrained_col)
+ table.append_constraint(
+ sa_schema.UniqueConstraint(*constrained_cols, name=conname))
@testing.requires.temp_table_reflection
def test_get_temp_table_unique_constraints(self):
insp = inspect(self.metadata.bind)
- eq_(
- insp.get_unique_constraints('user_tmp'),
- [{'column_names': ['name'], 'name': 'user_tmp_uq'}]
- )
+ reflected = insp.get_unique_constraints('user_tmp')
+ for refl in reflected:
+ # Different dialects handle duplicate index and constraints
+ # differently, so ignore this flag
+ refl.pop('duplicates_index', None)
+ eq_(reflected, [{'column_names': ['name'], 'name': 'user_tmp_uq'}])
@testing.requires.temp_table_reflection
def test_get_temp_table_indexes(self):
)
for orig, refl in zip(uniques, reflected):
+ # Different dialects handle duplicate index and constraints
+ # differently, so ignore this flag
+ refl.pop('duplicates_index', None)
eq_(orig, refl)
@testing.provide_metadata
view_names = dialect.get_view_names(connection, "information_schema")
self.assert_('TABLES' in view_names)
+ def test_reflection_with_unique_constraint(self):
+ insp = inspect(testing.db)
+
+ uc_table = Table('mysql_uc', MetaData(testing.db),
+ Column('a', String(10)),
+ UniqueConstraint('a', name='uc_a'))
+
+ try:
+ uc_table.create()
+
+ # MySQL converts unique constraints into unique indexes and
+ # the 0.9 API returns it as both an index and a constraint
+ indexes = set(i['name'] for i in insp.get_indexes('mysql_uc'))
+ constraints = set(i['name']
+ for i in insp.get_unique_constraints('mysql_uc'))
+
+ self.assert_('uc_a' in indexes)
+ self.assert_('uc_a' in constraints)
+
+ # However, upon creating a Table object via reflection, it should
+ # only appear as a unique index and not a constraint
+ reflected = Table('mysql_uc', MetaData(testing.db), autoload=True)
+
+ indexes = set(i.name for i in reflected.indexes)
+ constraints = set(uc.name for uc in reflected.constraints)
+
+ self.assert_('uc_a' in indexes)
+ self.assert_('uc_a' not in constraints)
+ finally:
+ uc_table.drop()
+
class RawReflectionTest(fixtures.TestBase):
def setup(self):
from sqlalchemy import testing
from sqlalchemy import inspect
from sqlalchemy import Table, Column, MetaData, Integer, String, \
- PrimaryKeyConstraint, ForeignKey, join, Sequence
+ PrimaryKeyConstraint, ForeignKey, join, Sequence, UniqueConstraint
from sqlalchemy import exc
import sqlalchemy as sa
from sqlalchemy.dialects.postgresql import base as postgresql
conn.execute("ALTER TABLE t RENAME COLUMN x to y")
ind = testing.db.dialect.get_indexes(conn, "t", None)
- eq_(ind, [{'unique': False, 'column_names': ['y'], 'name': 'idx1'}])
+ eq_(ind, [{'unique': False, 'duplicates_constraint': None,
+ 'column_names': ['y'], 'name': 'idx1'}])
conn.close()
@testing.provide_metadata
'labels': ['sad', 'ok', 'happy']
}])
+ def test_reflection_with_unique_constraint(self):
+ insp = inspect(testing.db)
+
+ uc_table = Table('pgsql_uc', MetaData(testing.db),
+ Column('a', String(10)),
+ UniqueConstraint('a', name='uc_a'))
+
+ try:
+ uc_table.create()
+
+ # PostgreSQL will create an implicit index for a unique
+ # constraint. As a result, the 0.9 API returns it as both
+ # an index and a constraint
+ indexes = set(i['name'] for i in insp.get_indexes('pgsql_uc'))
+ constraints = set(i['name']
+ for i in insp.get_unique_constraints('pgsql_uc'))
+
+ self.assert_('uc_a' in indexes)
+ self.assert_('uc_a' in constraints)
+
+ # However, upon creating a Table object via reflection, it should
+ # only appear as a unique constraint and not an index
+ reflected = Table('pgsql_uc', MetaData(testing.db), autoload=True)
+
+ indexes = set(i.name for i in reflected.indexes)
+ constraints = set(uc.name for uc in reflected.constraints)
+
+ self.assert_('uc_a' not in indexes)
+ self.assert_('uc_a' in constraints)
+ finally:
+ uc_table.drop()
+
class CustomTypeReflectionTest(fixtures.TestBase):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
