Skip to content

Remove unique constraints (fixes #100 and #104) #106

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Mar 4, 2022
Merged

Remove unique constraints (fixes #100 and #104) #106

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
bbf6a58
Add support for unique_constraint to introspection, and use it to det…
rubendv Feb 15, 2022
1e44d40
Also treat primary keys as constraints instead of as indexes.
rubendv Feb 28, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 9 additions & 1 deletion mssql/introspection.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -253,9 +253,14 @@ def get_constraints(self, cursor, table_name):
constraints[constraint] = {
"columns": [],
"primary_key": kind.lower() == "primary key",
# In the sys.indexes table, primary key indexes have is_unique_constraint as false,
# but is_unique as true.
"unique": kind.lower() in ["primary key", "unique"],
"unique_constraint": kind.lower() == "unique",
"foreign_key": (ref_table, ref_column) if kind.lower() == "foreign key" else None,
"check": False,
# Potentially misleading: primary key and unique constraints still have indexes attached to them.
# Should probably be updated with the additional info from the sys.indexes table we fetch later on.
"index": False,
}
# Record the details
Expand All @@ -280,6 +285,7 @@ def get_constraints(self, cursor, table_name):
"columns": [],
"primary_key": False,
"unique": False,
"unique_constraint": False,
"foreign_key": None,
"check": True,
"index": False,
Expand All @@ -291,6 +297,7 @@ def get_constraints(self, cursor, table_name):
SELECT
i.name AS index_name,
i.is_unique,
i.is_unique_constraint,
i.is_primary_key,
i.type,
i.type_desc,
Expand All @@ -316,12 +323,13 @@ def get_constraints(self, cursor, table_name):
ic.index_column_id ASC
""", [table_name])
indexes = {}
for index, unique, primary, type_, desc, order, column in cursor.fetchall():
for index, unique, unique_constraint, primary, type_, desc, order, column in cursor.fetchall():
if index not in indexes:
indexes[index] = {
"columns": [],
"primary_key": primary,
"unique": unique,
"unique_constraint": unique_constraint,
"foreign_key": None,
"check": False,
"index": True,
Expand Down
60 changes: 33 additions & 27 deletions mssql/schema.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -171,7 +171,12 @@ def alter_unique_together(self, model, old_unique_together, new_unique_together)
news = {tuple(fields) for fields in new_unique_together}
# Deleted uniques
for fields in olds.difference(news):
self._delete_composed_index(model, fields, {'unique': True}, self.sql_delete_index)
meta_constraint_names = {constraint.name for constraint in model._meta.constraints}
meta_index_names = {constraint.name for constraint in model._meta.indexes}
columns = [model._meta.get_field(field).column for field in fields]
self._delete_unique_constraint_for_columns(
model, columns, exclude=meta_constraint_names | meta_index_names, strict=True)

# Created uniques
if django_version >= (4, 0):
for field_names in news.difference(olds):
Expand Down Expand Up @@ -227,7 +232,7 @@ def _model_indexes_sql(self, model):

def _db_table_constraint_names(self, db_table, column_names=None, column_match_any=False,
unique=None, primary_key=None, index=None, foreign_key=None,
check=None, type_=None, exclude=None):
check=None, type_=None, exclude=None, unique_constraint=None):
"""
Return all constraint names matching the columns and conditions. Modified from base `_constraint_names`
`any_column_matches`=False: (default) only return constraints covering exactly `column_names`
Expand All @@ -247,6 +252,8 @@ def _db_table_constraint_names(self, db_table, column_names=None, column_match_a
):
if unique is not None and infodict['unique'] != unique:
continue
if unique_constraint is not None and infodict['unique_constraint'] != unique_constraint:
continue
if primary_key is not None and infodict['primary_key'] != primary_key:
continue
if index is not None and infodict['index'] != index:
Expand Down Expand Up @@ -299,16 +306,7 @@ def _alter_field(self, model, old_field, new_field, old_type, new_type,
self.execute(self._delete_constraint_sql(self.sql_delete_fk, model, fk_name))
# Has unique been removed?
if old_field.unique and (not new_field.unique or self._field_became_primary_key(old_field, new_field)):
# Find the unique constraint for this field
constraint_names = self._constraint_names(model, [old_field.column], unique=True, primary_key=False)
if strict and len(constraint_names) != 1:
raise ValueError("Found wrong number (%s) of unique constraints for %s.%s" % (
len(constraint_names),
model._meta.db_table,
old_field.column,
))
for constraint_name in constraint_names:
self.execute(self._delete_constraint_sql(self.sql_delete_unique, model, constraint_name))
self._delete_unique_constraint_for_columns(model, [old_field.column], strict=strict)
# Drop incoming FK constraints if the field is a primary key or unique,
# which might be a to_field target, and things are going to change.
drop_foreign_keys = (
Expand Down Expand Up @@ -694,21 +692,29 @@ def _delete_unique_constraints(self, model, old_field, new_field, strict=False):
unique_columns.append([old_field.column])
if unique_columns:
for columns in unique_columns:
constraint_names_normal = self._constraint_names(model, columns, unique=True, index=False)
constraint_names_index = self._constraint_names(model, columns, unique=True, index=True)
constraint_names = constraint_names_normal + constraint_names_index
if strict and len(constraint_names) != 1:
raise ValueError("Found wrong number (%s) of unique constraints for %s.%s" % (
len(constraint_names),
model._meta.db_table,
old_field.column,
))
for constraint_name in constraint_names_normal:
self.execute(self._delete_constraint_sql(self.sql_delete_unique, model, constraint_name))
# Unique indexes which are not table constraints must be deleted using the appropriate SQL.
# These may exist for example to enforce ANSI-compliant unique constraints on nullable columns.
for index_name in constraint_names_index:
self.execute(self._delete_constraint_sql(self.sql_delete_index, model, index_name))
self._delete_unique_constraint_for_columns(model, columns, strict=strict)

def _delete_unique_constraint_for_columns(self, model, columns, strict=False, **constraint_names_kwargs):
constraint_names_unique = self._db_table_constraint_names(
model._meta.db_table, columns, unique=True, unique_constraint=True, **constraint_names_kwargs)
constraint_names_primary = self._db_table_constraint_names(
model._meta.db_table, columns, unique=True, primary_key=True, **constraint_names_kwargs)
constraint_names_normal = constraint_names_unique + constraint_names_primary
constraint_names_index = self._db_table_constraint_names(
model._meta.db_table, columns, unique=True, unique_constraint=False, primary_key=False,
**constraint_names_kwargs)
constraint_names = constraint_names_normal + constraint_names_index
if strict and len(constraint_names) != 1:
raise ValueError("Found wrong number (%s) of unique constraints for columns %s" % (
len(constraint_names),
repr(columns),
))
for constraint_name in constraint_names_normal:
self.execute(self._delete_constraint_sql(self.sql_delete_unique, model, constraint_name))
# Unique indexes which are not table constraints must be deleted using the appropriate SQL.
# These may exist for example to enforce ANSI-compliant unique constraints on nullable columns.
for index_name in constraint_names_index:
self.execute(self._delete_constraint_sql(self.sql_delete_index, model, index_name))

def _rename_field_sql(self, table, old_field, new_field, new_type):
new_type = self._set_field_new_type_null_status(old_field, new_type)
Expand Down