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,24 @@ 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 _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 +241,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 +310,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 +463,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 +479,14 @@ 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]
+            index_columns.append(columns)
         if index_columns:
             for columns in index_columns:
                 index_names = self._constraint_names(model, columns, index=True)
@@ -605,7 +641,10 @@ def create_model(self, model):
         # 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))
+            condition = ' AND '.join(["[%s] IS NOT NULL" % col for col in columns])
+            sql_without_condition = self._create_unique_sql(model, columns)
+            sql_with_condition = self._create_unique_sql(model, columns, condition=condition)
+            self.deferred_sql.append(sql_with_condition)
         # 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,13 @@
 from django.db.models.expressions import Exists, OuterRef, Subquery
-from django.test import TestCase
+from django.test import TestCase, skipUnlessDBFeature
 
-from ..models import Comment, Post
+from ..models import Author, Comment, 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 +18,18 @@ 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)