Fixed #30060 -- Moved SQL generation for indexes and constraints to SchemaEditor.

8 files changed, 146 insertions, 117 deletions

diff --git a/django/db/backends/base/schema.py b/django/db/backends/base/schema.py
index 2a2c441f04..55835fe535 100644
--- a/django/db/backends/base/schema.py
+++ b/django/db/backends/base/schema.py
@@ -61,16 +61,21 @@ class BaseDatabaseSchemaEditor:
     sql_rename_column = "ALTER TABLE %(table)s RENAME COLUMN %(old_column)s TO %(new_column)s"
     sql_update_with_default = "UPDATE %(table)s SET %(column)s = %(default)s WHERE %(column)s IS NULL"
 
-    sql_foreign_key_constraint = "FOREIGN KEY (%(column)s) REFERENCES %(to_table)s (%(to_column)s)%(deferrable)s"
     sql_unique_constraint = "UNIQUE (%(columns)s)"
     sql_check_constraint = "CHECK (%(check)s)"
-    sql_create_constraint = "ALTER TABLE %(table)s ADD %(constraint)s"
     sql_delete_constraint = "ALTER TABLE %(table)s DROP CONSTRAINT %(name)s"
     sql_constraint = "CONSTRAINT %(name)s %(constraint)s"
 
-    sql_create_unique = None
+    sql_create_check = "ALTER TABLE %(table)s ADD CONSTRAINT %(name)s CHECK (%(check)s)"
+    sql_delete_check = sql_delete_constraint
+
+    sql_create_unique = "ALTER TABLE %(table)s ADD CONSTRAINT %(name)s UNIQUE (%(columns)s)"
     sql_delete_unique = sql_delete_constraint
 
+    sql_create_fk = (
+        "ALTER TABLE %(table)s ADD CONSTRAINT %(name)s FOREIGN KEY (%(column)s) "
+        "REFERENCES %(to_table)s (%(to_column)s)%(deferrable)s"
+    )
     sql_create_inline_fk = None
     sql_delete_fk = sql_delete_constraint
 
@@ -287,7 +292,7 @@ class BaseDatabaseSchemaEditor:
         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))
-        constraints = [check.full_constraint_sql(model, self) for check in model._meta.constraints]
+        constraints = [check.constraint_sql(model, self) for check in model._meta.constraints]
         # Make the table
         sql = self.sql_create_table % {
             "table": self.quote_name(model._meta.db_table),
@@ -422,7 +427,7 @@ class BaseDatabaseSchemaEditor:
         # Check constraints can go on the column SQL here
         db_params = field.db_parameters(connection=self.connection)
         if db_params['check']:
-            definition += " CHECK (%s)" % db_params['check']
+            definition += " " + self.sql_check_constraint % db_params
         # Build the SQL and run it
         sql = self.sql_create_column % {
             "table": self.quote_name(model._meta.db_table),
@@ -463,7 +468,7 @@ class BaseDatabaseSchemaEditor:
         if field.remote_field:
             fk_names = self._constraint_names(model, [field.column], foreign_key=True)
             for fk_name in fk_names:
-                self.execute(self._delete_constraint_sql(self.sql_delete_fk, model, fk_name))
+                self.execute(self._delete_fk_sql(model, fk_name))
         # Delete the column
         sql = self.sql_delete_column % {
             "table": self.quote_name(model._meta.db_table),
@@ -534,7 +539,7 @@ class BaseDatabaseSchemaEditor:
                 ))
             for fk_name in fk_names:
                 fks_dropped.add((old_field.column,))
-                self.execute(self._delete_constraint_sql(self.sql_delete_fk, model, fk_name))
+                self.execute(self._delete_fk_sql(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
@@ -546,7 +551,7 @@ class BaseDatabaseSchemaEditor:
                     old_field.column,
                 ))
             for constraint_name in constraint_names:
-                self.execute(self._delete_constraint_sql(self.sql_delete_unique, model, constraint_name))
+                self.execute(self._delete_unique_sql(model, constraint_name))
         # 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 = (
@@ -563,7 +568,7 @@ class BaseDatabaseSchemaEditor:
                     new_rel.related_model, [new_rel.field.column], foreign_key=True
                 )
                 for fk_name in rel_fk_names:
-                    self.execute(self._delete_constraint_sql(self.sql_delete_fk, new_rel.related_model, fk_name))
+                    self.execute(self._delete_fk_sql(new_rel.related_model, fk_name))
         # Removed an index? (no strict check, as multiple indexes are possible)
         # Remove indexes if db_index switched to False or a unique constraint
         # will now be used in lieu of an index. The following lines from the
@@ -585,7 +590,7 @@ class BaseDatabaseSchemaEditor:
                     # The only way to check if an index was created with
                     # db_index=True or with Index(['field'], name='foo')
                     # is to look at its name (refs #28053).
-                    self.execute(self._delete_constraint_sql(self.sql_delete_index, model, index_name))
+                    self.execute(self._delete_index_sql(model, index_name))
         # Change check constraints?
         if old_db_params['check'] != new_db_params['check'] and old_db_params['check']:
             constraint_names = self._constraint_names(model, [old_field.column], check=True)
@@ -596,7 +601,7 @@ class BaseDatabaseSchemaEditor:
                     old_field.column,
                 ))
             for constraint_name in constraint_names:
-                self.execute(self._delete_constraint_sql(self.sql_delete_constraint, model, constraint_name))
+                self.execute(self._delete_check_sql(model, constraint_name))
         # Have they renamed the column?
         if old_field.column != new_field.column:
             self.execute(self._rename_field_sql(model._meta.db_table, old_field, new_field, new_type))
@@ -707,15 +712,7 @@ class BaseDatabaseSchemaEditor:
         # Changed to become primary key?
         if self._field_became_primary_key(old_field, new_field):
             # Make the new one
-            self.execute(
-                self.sql_create_pk % {
-                    "table": self.quote_name(model._meta.db_table),
-                    "name": self.quote_name(
-                        self._create_index_name(model._meta.db_table, [new_field.column], suffix="_pk")
-                    ),
-                    "columns": self.quote_name(new_field.column),
-                }
-            )
+            self.execute(self._create_primary_key_sql(model, new_field))
             # Update all referencing columns
             rels_to_update.extend(_related_non_m2m_objects(old_field, new_field))
         # Handle our type alters on the other end of rels from the PK stuff above
@@ -746,18 +743,8 @@ class BaseDatabaseSchemaEditor:
                     self.execute(self._create_fk_sql(rel.related_model, rel.field, "_fk"))
         # Does it have check constraints we need to add?
         if old_db_params['check'] != new_db_params['check'] and new_db_params['check']:
-            constraint = self.sql_constraint % {
-                'name': self.quote_name(
-                    self._create_index_name(model._meta.db_table, [new_field.column], suffix='_check'),
-                ),
-                'constraint': self.sql_check_constraint % new_db_params,
-            }
-            self.execute(
-                self.sql_create_constraint % {
-                    'table': self.quote_name(model._meta.db_table),
-                    'constraint': constraint,
-                }
-            )
+            constraint_name = self._create_index_name(model._meta.db_table, [new_field.column], suffix='_check')
+            self.execute(self._create_check_sql(model, constraint_name, new_db_params['check']))
         # Drop the default if we need to
         # (Django usually does not use in-database defaults)
         if needs_database_default:
@@ -931,7 +918,14 @@ class BaseDatabaseSchemaEditor:
             using=using,
             columns=self._index_columns(table, columns, col_suffixes, opclasses),
             extra=tablespace_sql,
-            condition=condition,
+            condition=(' WHERE ' + condition) if condition else '',
+        )
+
+    def _delete_index_sql(self, model, name):
+        return Statement(
+            self.sql_delete_index,
+            table=Table(model._meta.db_table, self.quote_name),
+            name=self.quote_name(name),
         )
 
     def _index_columns(self, table, columns, col_suffixes, opclasses):
@@ -984,10 +978,6 @@ class BaseDatabaseSchemaEditor:
             "type": new_type,
         }
 
-    def _create_constraint_sql(self, table, name, constraint):
-        constraint = Statement(self.sql_constraint, name=name, constraint=constraint)
-        return Statement(self.sql_create_constraint, table=table, constraint=constraint)
-
     def _create_fk_sql(self, model, field, suffix):
         def create_fk_name(*args, **kwargs):
             return self.quote_name(self._create_index_name(*args, **kwargs))
@@ -1005,14 +995,27 @@ class BaseDatabaseSchemaEditor:
         to_table = Table(field.target_field.model._meta.db_table, self.quote_name)
         to_column = Columns(field.target_field.model._meta.db_table, [field.target_field.column], self.quote_name)
         deferrable = self.connection.ops.deferrable_sql()
-        constraint = Statement(
-            self.sql_foreign_key_constraint,
+        return Statement(
+            self.sql_create_fk,
+            table=table,
+            name=name,
             column=column,
             to_table=to_table,
             to_column=to_column,
             deferrable=deferrable,
         )
-        return self._create_constraint_sql(table, name, constraint)
+
+    def _delete_fk_sql(self, model, name):
+        return self._delete_constraint_sql(self.sql_delete_fk, model, name)
+
+    def _unique_sql(self, fields, name):
+        constraint = self.sql_unique_constraint % {
+            'columns': ', '.join(map(self.quote_name, fields)),
+        }
+        return self.sql_constraint % {
+            'name': self.quote_name(name),
+            'constraint': constraint,
+        }
 
     def _create_unique_sql(self, model, columns, name=None):
         def create_unique_name(*args, **kwargs):
@@ -1024,24 +1027,40 @@ class BaseDatabaseSchemaEditor:
         else:
             name = self.quote_name(name)
         columns = Columns(table, columns, self.quote_name)
-        if self.sql_create_unique:
-            # Some databases use a different syntax for unique constraint
-            # creation.
-            return Statement(
-                self.sql_create_unique,
-                table=table,
-                name=name,
-                columns=columns,
-            )
-        constraint = Statement(self.sql_unique_constraint, columns=columns)
-        return self._create_constraint_sql(table, name, constraint)
+        return Statement(
+            self.sql_create_unique,
+            table=table,
+            name=name,
+            columns=columns,
+        )
 
-    def _delete_constraint_sql(self, template, model, name):
-        return template % {
-            "table": self.quote_name(model._meta.db_table),
-            "name": self.quote_name(name),
+    def _delete_unique_sql(self, model, name):
+        return self._delete_constraint_sql(self.sql_delete_unique, model, name)
+
+    def _check_sql(self, name, check):
+        return self.sql_constraint % {
+            'name': self.quote_name(name),
+            'constraint': self.sql_check_constraint % {'check': check},
         }
 
+    def _create_check_sql(self, model, name, check):
+        return Statement(
+            self.sql_create_check,
+            table=Table(model._meta.db_table, self.quote_name),
+            name=self.quote_name(name),
+            check=check,
+        )
+
+    def _delete_check_sql(self, model, name):
+        return self._delete_constraint_sql(self.sql_delete_check, model, name)
+
+    def _delete_constraint_sql(self, template, model, name):
+        return Statement(
+            template,
+            table=Table(model._meta.db_table, self.quote_name),
+            name=self.quote_name(name),
+        )
+
     def _constraint_names(self, model, column_names=None, unique=None,
                           primary_key=None, index=None, foreign_key=None,
                           check=None, type_=None):
@@ -1079,7 +1098,20 @@ class BaseDatabaseSchemaEditor:
                 model._meta.db_table,
             ))
         for constraint_name in constraint_names:
-            self.execute(self._delete_constraint_sql(self.sql_delete_pk, model, constraint_name))
+            self.execute(self._delete_primary_key_sql(model, constraint_name))
+
+    def _create_primary_key_sql(self, model, field):
+        return Statement(
+            self.sql_create_pk,
+            table=Table(model._meta.db_table, self.quote_name),
+            name=self.quote_name(
+                self._create_index_name(model._meta.db_table, [field.column], suffix="_pk")
+            ),
+            columns=Columns(model._meta.db_table, [field.column], self.quote_name),
+        )
+
+    def _delete_primary_key_sql(self, model, name):
+        return self._delete_constraint_sql(self.sql_delete_pk, model, name)
 
     def remove_procedure(self, procedure_name, param_types=()):
         sql = self.sql_delete_procedure % {
diff --git a/django/db/backends/postgresql/schema.py b/django/db/backends/postgresql/schema.py
index dc9959ab4f..9eecaac19a 100644
--- a/django/db/backends/postgresql/schema.py
+++ b/django/db/backends/postgresql/schema.py
@@ -114,7 +114,7 @@ class DatabaseSchemaEditor(BaseDatabaseSchemaEditor):
             (old_type.startswith('citext') and not new_type.startswith('citext'))
         ):
             index_name = self._create_index_name(model._meta.db_table, [old_field.column], suffix='_like')
-            self.execute(self._delete_constraint_sql(self.sql_delete_index, model, index_name))
+            self.execute(self._delete_index_sql(model, index_name))
 
         super()._alter_field(
             model, old_field, new_field, old_type, new_type, old_db_params,
@@ -130,7 +130,7 @@ class DatabaseSchemaEditor(BaseDatabaseSchemaEditor):
         # Removed an index? Drop any PostgreSQL-specific indexes.
         if old_field.unique and not (new_field.db_index or new_field.unique):
             index_to_remove = self._create_index_name(model._meta.db_table, [old_field.column], suffix='_like')
-            self.execute(self._delete_constraint_sql(self.sql_delete_index, model, index_to_remove))
+            self.execute(self._delete_index_sql(model, index_to_remove))
 
     def _index_columns(self, table, columns, col_suffixes, opclasses):
         if opclasses:
diff --git a/django/db/backends/sqlite3/schema.py b/django/db/backends/sqlite3/schema.py
index 1a4b94fa82..e8e40bbe34 100644
--- a/django/db/backends/sqlite3/schema.py
+++ b/django/db/backends/sqlite3/schema.py
@@ -11,10 +11,10 @@ from django.db.utils import NotSupportedError
 class DatabaseSchemaEditor(BaseDatabaseSchemaEditor):
 
     sql_delete_table = "DROP TABLE %(table)s"
+    sql_create_fk = None
     sql_create_inline_fk = "REFERENCES %(to_table)s (%(to_column)s) DEFERRABLE INITIALLY DEFERRED"
     sql_create_unique = "CREATE UNIQUE INDEX %(name)s ON %(table)s (%(columns)s)"
     sql_delete_unique = "DROP INDEX %(name)s"
-    sql_foreign_key_constraint = None
 
     def __enter__(self):
         # Some SQLite schema alterations need foreign key constraints to be
diff --git a/django/db/models/constraints.py b/django/db/models/constraints.py
index f6ea0fa12a..5b6f498d94 100644
--- a/django/db/models/constraints.py
+++ b/django/db/models/constraints.py
@@ -10,25 +10,11 @@ class BaseConstraint:
     def constraint_sql(self, model, schema_editor):
         raise NotImplementedError('This method must be implemented by a subclass.')
 
-    def full_constraint_sql(self, model, schema_editor):
-        return schema_editor.sql_constraint % {
-            'name': schema_editor.quote_name(self.name),
-            'constraint': self.constraint_sql(model, schema_editor),
-        }
-
     def create_sql(self, model, schema_editor):
-        sql = self.full_constraint_sql(model, schema_editor)
-        return schema_editor.sql_create_constraint % {
-            'table': schema_editor.quote_name(model._meta.db_table),
-            'constraint': sql,
-        }
+        raise NotImplementedError('This method must be implemented by a subclass.')
 
     def remove_sql(self, model, schema_editor):
-        quote_name = schema_editor.quote_name
-        return schema_editor.sql_delete_constraint % {
-            'table': quote_name(model._meta.db_table),
-            'name': quote_name(self.name),
-        }
+        raise NotImplementedError('This method must be implemented by a subclass.')
 
     def deconstruct(self):
         path = '%s.%s' % (self.__class__.__module__, self.__class__.__name__)
@@ -45,14 +31,23 @@ class CheckConstraint(BaseConstraint):
         self.check = check
         super().__init__(name)
 
-    def constraint_sql(self, model, schema_editor):
-        query = Query(model)
+    def _get_check_sql(self, model, schema_editor):
+        query = Query(model=model)
         where = query.build_where(self.check)
-        connection = schema_editor.connection
-        compiler = connection.ops.compiler('SQLCompiler')(query, connection, 'default')
-        sql, params = where.as_sql(compiler, connection)
-        params = tuple(schema_editor.quote_value(p) for p in params)
-        return schema_editor.sql_check_constraint % {'check': sql % params}
+        compiler = query.get_compiler(connection=schema_editor.connection)
+        sql, params = where.as_sql(compiler, schema_editor.connection)
+        return sql % tuple(schema_editor.quote_value(p) for p in params)
+
+    def constraint_sql(self, model, schema_editor):
+        check = self._get_check_sql(model, schema_editor)
+        return schema_editor._check_sql(self.name, check)
+
+    def create_sql(self, model, schema_editor):
+        check = self._get_check_sql(model, schema_editor)
+        return schema_editor._create_check_sql(model, self.name, check)
+
+    def remove_sql(self, model, schema_editor):
+        return schema_editor._delete_check_sql(model, self.name)
 
     def __repr__(self):
         return "<%s: check='%s' name=%r>" % (self.__class__.__name__, self.check, self.name)
@@ -78,17 +73,15 @@ class UniqueConstraint(BaseConstraint):
         super().__init__(name)
 
     def constraint_sql(self, model, schema_editor):
-        columns = (
-            model._meta.get_field(field_name).column
-            for field_name in self.fields
-        )
-        return schema_editor.sql_unique_constraint % {
-            'columns': ', '.join(map(schema_editor.quote_name, columns)),
-        }
+        fields = [model._meta.get_field(field_name).column for field_name in self.fields]
+        return schema_editor._unique_sql(fields, self.name)
 
     def create_sql(self, model, schema_editor):
-        columns = [model._meta.get_field(field_name).column for field_name in self.fields]
-        return schema_editor._create_unique_sql(model, columns, self.name)
+        fields = [model._meta.get_field(field_name).column for field_name in self.fields]
+        return schema_editor._create_unique_sql(model, fields, self.name)
+
+    def remove_sql(self, model, schema_editor):
+        return schema_editor._delete_unique_sql(model, self.name)
 
     def __repr__(self):
         return '<%s: fields=%r name=%r>' % (self.__class__.__name__, self.fields, self.name)
diff --git a/django/db/models/indexes.py b/django/db/models/indexes.py
index 48fc74ad64..6bf8803e8b 100644
--- a/django/db/models/indexes.py
+++ b/django/db/models/indexes.py
@@ -65,7 +65,7 @@ class Index:
         sql, params = query.where.as_sql(compiler=compiler, connection=schema_editor.connection)
         # BaseDatabaseSchemaEditor does the same map on the params, but since
         # it's handled outside of that class, the work is done here.
-        return ' WHERE ' + (sql % tuple(map(schema_editor.quote_value, params)))
+        return sql % tuple(map(schema_editor.quote_value, params))
 
     def create_sql(self, model, schema_editor, using=''):
         fields = [model._meta.get_field(field_name) for field_name, _ in self.fields_orders]
@@ -77,11 +77,7 @@ class Index:
         )
 
     def remove_sql(self, model, schema_editor):
-        quote_name = schema_editor.quote_name
-        return schema_editor.sql_delete_index % {
-            'table': quote_name(model._meta.db_table),
-            'name': quote_name(self.name),
-        }
+        return schema_editor._delete_index_sql(model, self.name)
 
     def deconstruct(self):
         path = '%s.%s' % (self.__class__.__module__, self.__class__.__name__)
diff --git a/docs/releases/2.2.txt b/docs/releases/2.2.txt
index 3201d00047..65f53c1a49 100644
--- a/docs/releases/2.2.txt
+++ b/docs/releases/2.2.txt
@@ -322,13 +322,6 @@ backends.
 * Third party database backends must implement support for partial indexes or
   set ``DatabaseFeatures.supports_partial_indexes`` to ``False``.
 
-* Several ``SchemaEditor`` attributes are changed:
-
-  * ``sql_create_check`` is replaced with ``sql_create_constraint``.
-  * ``sql_delete_check`` is replaced with ``sql_delete_constraint``.
-  * ``sql_create_fk`` is replaced with ``sql_foreign_key_constraint``,
-    ``sql_constraint``, and ``sql_create_constraint``.
-
 * ``DatabaseIntrospection.table_name_converter()`` and
   ``column_name_converter()`` are removed. Third party database backends may
   need to instead implement ``DatabaseIntrospection.identifier_converter()``.
@@ -336,6 +329,15 @@ backends.
   ``DatabaseIntrospection.get_constraints()`` returns must be normalized by
   ``identifier_converter()``.
 
+* SQL generation for indexes is moved from :class:`~django.db.models.Index` to
+  ``SchemaEditor`` and these ``SchemaEditor`` methods are added:
+
+  * ``_create_primary_key_sql()`` and ``_delete_primary_key_sql()``
+  * ``_delete_index_sql()`` (to pair with ``_create_index_sql()``)
+  * ``_delete_unique_sql`` (to pair with ``_create_unique_sql()``)
+  * ``_delete_fk_sql()`` (to pair with ``_create_fk_sql()``)
+  * ``_create_check_sql()`` and ``_delete_check_sql()``
+
 Admin actions are no longer collected from base ``ModelAdmin`` classes
 ----------------------------------------------------------------------
 
diff --git a/tests/constraints/tests.py b/tests/constraints/tests.py
index d53892c3a3..995c5f461c 100644
--- a/tests/constraints/tests.py
+++ b/tests/constraints/tests.py
@@ -18,6 +18,18 @@ class BaseConstraintTests(SimpleTestCase):
         with self.assertRaisesMessage(NotImplementedError, msg):
             c.constraint_sql(None, None)
 
+    def test_create_sql(self):
+        c = BaseConstraint('name')
+        msg = 'This method must be implemented by a subclass.'
+        with self.assertRaisesMessage(NotImplementedError, msg):
+            c.create_sql(None, None)
+
+    def test_remove_sql(self):
+        c = BaseConstraint('name')
+        msg = 'This method must be implemented by a subclass.'
+        with self.assertRaisesMessage(NotImplementedError, msg):
+            c.remove_sql(None, None)
+
 
 class CheckConstraintTests(TestCase):
     def test_repr(self):
diff --git a/tests/schema/tests.py b/tests/schema/tests.py
index 361a5a9466..410a52b646 100644
--- a/tests/schema/tests.py
+++ b/tests/schema/tests.py
@@ -2147,23 +2147,17 @@ class SchemaTests(TransactionTestCase):
             editor.alter_field(model, get_field(unique=True), field, strict=True)
             self.assertNotIn(expected_constraint_name, self.get_constraints(model._meta.db_table))
 
-            if editor.sql_foreign_key_constraint:
+            if editor.sql_create_fk:
                 constraint_name = 'CamelCaseFKConstraint'
                 expected_constraint_name = identifier_converter(constraint_name)
-                fk_sql = editor.sql_foreign_key_constraint % {
-                    "column": editor.quote_name(column),
-                    "to_table": editor.quote_name(table),
-                    "to_column": editor.quote_name(model._meta.auto_field.column),
-                    "deferrable": connection.ops.deferrable_sql(),
-                }
-                constraint_sql = editor.sql_constraint % {
-                    "name": editor.quote_name(constraint_name),
-                    "constraint": fk_sql,
-                }
                 editor.execute(
-                    editor.sql_create_constraint % {
+                    editor.sql_create_fk % {
                         "table": editor.quote_name(table),
-                        "constraint": constraint_sql,
+                        "name": editor.quote_name(constraint_name),
+                        "column": editor.quote_name(column),
+                        "to_table": editor.quote_name(table),
+                        "to_column": editor.quote_name(model._meta.auto_field.column),
+                        "deferrable": connection.ops.deferrable_sql(),
                     }
                 )
                 self.assertIn(expected_constraint_name, self.get_constraints(model._meta.db_table))