Add support for nullable fields in unique_together

Author: OskarPersson

setup.cfg

@@ -1,5 +1,5 @@
 [flake8]
-exclude = .git,__pycache__,
+exclude = .git,__pycache__,migrations
 # W504 is mutually exclusive with W503
 ignore = W504
 max-line-length = 119

sql_server/pyodbc/features.py

@@ -23,7 +23,6 @@ class DatabaseFeatures(BaseDatabaseFeatures):
     supports_ignore_conflicts = False
     supports_index_on_text_field = False
     supports_paramstyle_pyformat = False
-    supports_partially_nullable_unique_constraints = False
     supports_regex_backreferencing = False
     supports_sequence_reset = False
     supports_subqueries_in_group_by = False
@@ -41,6 +40,10 @@ def has_bulk_insert(self):
     def supports_nullable_unique_constraints(self):
         return self.connection.sql_server_version > 2005
 
+    @cached_property
+    def supports_partially_nullable_unique_constraints(self):
+        return self.connection.sql_server_version > 2005
+
     @cached_property
     def supports_partial_indexes(self):
         return self.connection.sql_server_version > 2005

sql_server/pyodbc/schema.py

@@ -123,6 +123,70 @@ def _alter_column_type_sql(self, model, old_field, new_field, new_type):
         new_type = self._set_field_new_type_null_status(old_field, new_type)
         return super()._alter_column_type_sql(model, old_field, new_field, new_type)
 
+    def alter_unique_together(self, model, old_unique_together, new_unique_together):
+        """
+        Deal with a model changing its unique_together. The input
+        unique_togethers must be doubly-nested, not the single-nested
+        ["foo", "bar"] format.
+        """
+        olds = {tuple(fields) for fields in old_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)
+        # Created uniques
+        for fields in news.difference(olds):
+            columns = [model._meta.get_field(field).column for field in fields]
+            condition = ' AND '.join(["[%s] IS NOT NULL" % col for col in columns])
+            sql = self._create_unique_sql(model, columns, condition=condition)
+            self.execute(sql)
+
+    def _model_indexes_sql(self, model):
+        """
+        Return a list of all index SQL statements (field indexes,
+        index_together, Meta.indexes) for the specified model.
+        """
+        if not model._meta.managed or model._meta.proxy or model._meta.swapped:
+            return []
+        output = []
+        for field in model._meta.local_fields:
+            output.extend(self._field_indexes_sql(model, field))
+
+        for field_names in model._meta.index_together:
+            fields = [model._meta.get_field(field) for field in field_names]
+            output.append(self._create_index_sql(model, fields, suffix="_idx"))
+
+        for field_names in model._meta.unique_together:
+            columns = [model._meta.get_field(field).column for field in field_names]
+            condition = ' AND '.join(["[%s] IS NOT NULL" % col for col in columns])
+            sql = self._create_unique_sql(model, columns, condition=condition)
+            output.append(sql)
+
+        for index in model._meta.indexes:
+            output.append(index.create_sql(model, self))
+        return output
+
+    def _alter_many_to_many(self, model, old_field, new_field, strict):
+        """Alter M2Ms to repoint their to= endpoints."""
+
+        for idx in self._constraint_names(old_field.remote_field.through, index=True, unique=True):
+            self.execute(self.sql_delete_index % {'name': idx, 'table': old_field.remote_field.through._meta.db_table})
+
+        return super()._alter_many_to_many(model, old_field, new_field, strict)
+
+    def alter_db_table(self, model, old_db_table, new_db_table):
+        index_names = self._constraint_names(model, index=True)
+        for index_name in index_names:
+            self.execute(self._delete_constraint_sql(self.sql_delete_index, model, index_name))
+
+        model._meta.db_table = old_db_table
+        index_names = self._constraint_names(model, index=True)
+        for index_name in index_names:
+            self.execute(self._delete_constraint_sql(self.sql_delete_index, model, index_name))
+        model._meta.db_table = new_db_table
+
+        return super().alter_db_table(model, old_db_table, new_db_table)
+
     def _alter_field(self, model, old_field, new_field, old_type, new_type,
                      old_db_params, new_db_params, strict=False):
         """Actually perform a "physical" (non-ManyToMany) field update."""
@@ -223,12 +287,18 @@ def _alter_field(self, model, old_field, new_field, old_type, new_type,
             for constraint_name in constraint_names:
                 self.execute(self._delete_constraint_sql(self.sql_delete_check, model, constraint_name))
         # Have they renamed the column?
+        old_create_indexes_sql = []
         if old_field.column != new_field.column:
+            # remove old indices
+            old_create_indexes_sql = self._model_indexes_sql(model)
+            self._delete_indexes(model, old_field, new_field)
+
             self.execute(self._rename_field_sql(model._meta.db_table, old_field, new_field, new_type))
             # Rename all references to the renamed column.
             for sql in self.deferred_sql:
                 if isinstance(sql, Statement):
                     sql.rename_column_references(model._meta.db_table, old_field.column, new_field.column)
+
         # Next, start accumulating actions to do
         actions = []
         null_actions = []
@@ -286,6 +356,7 @@ def _alter_field(self, model, old_field, new_field, old_type, new_type,
                 actions = [(", ".join(sql), sum(params, []))]
             # Apply those actions
             for sql, params in actions:
+                self._delete_indexes(model, old_field, new_field)
                 self.execute(
                     self.sql_alter_column % {
                         "table": self.quote_name(model._meta.db_table),
@@ -438,6 +509,14 @@ def _alter_field(self, model, old_field, new_field, old_type, new_type,
                 "changes": changes_sql,
             }
             self.execute(sql, params)
+
+        for o in old_create_indexes_sql:
+            o = str(o)
+            if old_field.column not in o:
+                continue
+            o = o.replace('[%s]' % old_field.column, '[%s]' % new_field.column)
+            self.execute(o)
+
         # Reset connection if required
         if self.connection.features.connection_persists_old_columns:
             self.connection.close()
@@ -446,11 +525,15 @@ def _delete_indexes(self, model, old_field, new_field):
         index_columns = []
         if old_field.db_index and new_field.db_index:
             index_columns.append([old_field.column])
-        else:
-            for fields in model._meta.index_together:
-                columns = [model._meta.get_field(field).column for field in fields]
-                if old_field.column in columns:
-                    index_columns.append(columns)
+        for fields in model._meta.index_together:
+            columns = [model._meta.get_field(field).column for field in fields]
+            if old_field.column in columns:
+                index_columns.append(columns)
+
+        for fields in model._meta.unique_together:
+            columns = [model._meta.get_field(field).column for field in fields]
+            if old_field.column in columns:
+                index_columns.append(columns)
         if index_columns:
             for columns in index_columns:
                 index_names = self._constraint_names(model, columns, index=True)
@@ -601,11 +684,6 @@ def create_model(self, model):
                 if autoinc_sql:
                     self.deferred_sql.extend(autoinc_sql)
 
-        # Add any unique_togethers (always deferred, as some fields might be
-        # created afterwards, like geometry fields with some backends)
-        for fields in model._meta.unique_together:
-            columns = [model._meta.get_field(field).column for field in fields]
-            self.deferred_sql.append(self._create_unique_sql(model, columns))
         # Make the table
         sql = self.sql_create_table % {
             "table": self.quote_name(model._meta.db_table),

testapp/migrations/0001_initial.py

@@ -14,13 +14,41 @@ class Migration(migrations.Migration):
     ]
 
     operations = [
+        migrations.CreateModel(
+            name='Author',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('name', models.CharField(max_length=100)),
+            ],
+        ),
+        migrations.CreateModel(
+            name='Editor',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('name', models.CharField(max_length=100)),
+            ],
+        ),
         migrations.CreateModel(
             name='Post',
             fields=[
                 ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                 ('title', models.CharField(max_length=255, verbose_name='title')),
             ],
         ),
+        migrations.AddField(
+            model_name='post',
+            name='alt_editor',
+            field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, to='testapp.Editor'),
+        ),
+        migrations.AddField(
+            model_name='post',
+            name='author',
+            field=models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='testapp.Author'),
+        ),
+        migrations.AlterUniqueTogether(
+            name='post',
+            unique_together={('author', 'title', 'alt_editor')},
+        ),
         migrations.CreateModel(
             name='Comment',
             fields=[

testapp/models.py

@@ -4,8 +4,24 @@
 from django.utils import timezone
 
 
+class Author(models.Model):
+    name = models.CharField(max_length=100)
+
+
+class Editor(models.Model):
+    name = models.CharField(max_length=100)
+
+
 class Post(models.Model):
     title = models.CharField('title', max_length=255)
+    author = models.ForeignKey(Author, models.CASCADE)
+    # Optional secondary author
+    alt_editor = models.ForeignKey(Editor, models.SET_NULL, blank=True, null=True)
+
+    class Meta:
+        unique_together = (
+            ('author', 'title', 'alt_editor'),
+        )
 
     def __str__(self):
         return self.title

testapp/tests/test_expressions.py

@@ -1,12 +1,14 @@
 from django.db.models.expressions import Exists, OuterRef, Subquery
-from django.test import TestCase
+from django.db.utils import IntegrityError
+from django.test import TestCase, skipUnlessDBFeature
 
-from ..models import Comment, Post
+from ..models import Author, Comment, Editor, Post
 
 
 class TestSubquery(TestCase):
     def setUp(self):
-        self.post = Post.objects.create(title="foo")
+        self.author = Author.objects.create(name="author")
+        self.post = Post.objects.create(title="foo", author=self.author)
 
     def test_with_count(self):
         newest = Comment.objects.filter(post=OuterRef('pk')).order_by('-created_at')
@@ -17,9 +19,23 @@ def test_with_count(self):
 
 class TestExists(TestCase):
     def setUp(self):
-        self.post = Post.objects.create(title="foo")
+        self.author = Author.objects.create(name="author")
+        self.post = Post.objects.create(title="foo", author=self.author)
 
     def test_with_count(self):
         Post.objects.annotate(
             post_exists=Exists(Post.objects.all())
         ).filter(post_exists=True).count()
+
+
+@skipUnlessDBFeature('supports_partially_nullable_unique_constraints')
+class TestPartiallyNullableUniqueTogether(TestCase):
+    def test_partially_nullable(self):
+        author = Author.objects.create(name="author")
+        Post.objects.create(title="foo", author=author)
+        Post.objects.create(title="foo", author=author)
+
+        editor = Editor.objects.create(name="editor")
+        Post.objects.create(title="foo", author=author, alt_editor=editor)
+        with self.assertRaises(IntegrityError):
+            Post.objects.create(title="foo", author=author, alt_editor=editor)