when altering the given field must constraints on its model from the given relation be temporarily dropped m2m reverse field foreign key constraint on the primary key which is being altered is the constraint targeting the field being altered related fields must be returned in a deterministic order filter out m2m objects from reverse relations return oldrelation newrelation tuples this class and its subclasses are responsible for emitting schemachanging statements to the databases model creationremovalalteration field renaming index fiddling and so on overrideable sql templates statemanaging methods core utility functions execute the given sql statement with optional parameters don t perform the transactional ddl check if sql is being collected as it s not going to be executed anyway if not self collectsql and self connection inatomicblock and not self connection features canrollbackddl raise transactionmanagementerror executing ddl statements while in a transaction on databases that can t perform a rollback is prohibited account for nonstring statement objects sql strsql log the command we re running then run it logger debug s params r sql params extraparams params sql sql if self collectsql ending if sql rstrip endswith else if params is not none self collectedsql append sql tuplemapself quotevalue params ending else self collectedsql appendsql ending else with self connection cursor as cursor cursor executesql params def quotenameself name return self connection ops quotenamename def tablesqlself model add any uniquetogethers always deferred as some fields might be created afterward like geometry fields with some backends create column sql add fk deferreds if needed sql check constraints can go on the column sql here autoincrement sql for backends with inline variant fk add the sql to our big list autoincrement sql for backends with post table definition variant field database mapping functions work out nullability add database default include a default value if requested don t include a default value if it s a nullable field and the default cannot be dropped in the alter column statement e g mysql longtext and longblob some databases can t take defaults as a parameter oracle if this is the case the individual schema backend should implement preparedefault oracle treats the empty string as null so coerce the null option whenever is a possible value optionally add the tablespace if it s an implicitly indexed column return the column definition for a field the field must already have had setattributesfromname called get the column s type and use that as the basis of the sql check for fields that aren t actually columns e g m2m this appends to the params being returned some backends don t accept default values for certain columns types i e mysql longtext and longblob some backends don t accept default values for certain columns types i e mysql longtext and longblob in the alter column statement only used for backends which have requiresliteraldefaults feature return the sql to use in a default clause the resulting string should contain a s placeholder for a default value return the sql and params for the field s database default from django db models expressions import value sql s if isinstancefield dbdefault value else s query querymodelfield model compiler query getcompilerconnectionself connection defaultsql params compiler compilefield dbdefault if self connection features requiresliteraldefaults some databases doesn t support parameterized defaults oracle sqlite if this is the case the individual schema backend should implement preparedefault defaultsql tupleself preparedefaultp for p in params params return sql defaultsql params def columngeneratedsqlself field this method allows testing its logic without a connection return a field s effective database default value return field getdbprepsaveself effectivedefaultfield self connection def quotevalueself value raise notimplementederror actions def createmodelself model sql params self tablesqlmodel prevent using as params in the case a literal is used in the definition self executesql params or none if self connection features supportscomments add table comment if model meta dbtablecomment self alterdbtablecommentmodel none model meta dbtablecomment add column comments if not self connection features supportscommentsinline for field in model meta localfields if field dbcomment fielddbparams field dbparameters connectionself connection fieldtype fielddbparamstype self execute self altercolumncommentsql model field fieldtype field dbcomment add any field index deferred as sqlite remaketable needs it self deferredsql extendself modelindexessqlmodel make m2m tables for field in model meta localmanytomany if field remotefield through meta autocreated self createmodelfield remotefield through def deletemodelself model handle autocreated intermediary models delete the table remove all deferred statements referencing the deleted table add an index on a model if index containsexpressions and not self connection features supportsexpressionindexes return none index createsql returns interpolated sql which makes paramsnone a necessity to avoid escaping attempts on execution self executeindex createsqlmodel self paramsnone def removeindexself model index add a constraint to a model sql constraint createsqlmodel self if sql constraint createsql returns interpolated sql which makes paramsnone a necessity to avoid escaping attempts on execution self executesql paramsnone def removeconstraintself model constraint deal with a model changing its uniquetogether the input uniquetogethers must be doublynested not the singlenested foo bar format deleted uniques created uniques deal with a model changing its indextogether the input indextogethers must be doublynested not the singlenested foo bar format deleted indexes created indexes constraint matching the uniquetogether name rename the table a model points to if olddbtable newdbtable or self connection features ignorestablenamecase and olddbtable lower newdbtable lower return self execute self sqlrenametable oldtable self quotenameolddbtable newtable self quotenamenewdbtable rename all references to the old table name for sql in self deferredsql if isinstancesql statement sql renametablereferencesolddbtable newdbtable def alterdbtablecommentself model olddbtablecomment newdbtablecomment if self sqlaltertablecomment and self connection features supportscomments self execute self sqlaltertablecomment table self quotenamemodel meta dbtable comment self quotevaluenewdbtablecomment or def alterdbtablespaceself model olddbtablespace newdbtablespace create a field on a model usually involves adding a column but may involve adding a table instead for m2m fields specialcase implicit m2m tables get the column s definition it might not actually have a column behind it check constraints can go on the column sql here add fk constraint inline if supported otherwise add fk constraints later build the sql and run it drop the default if we need to add field comment if required add an index if required reset connection if required remove a field from a model usually involves deleting a column but for m2ms may involve deleting a table specialcase implicit m2m tables it might not actually have a column behind it drop any fk constraints mysql requires explicit deletion delete the column reset connection if required remove all deferred statements referencing the deleted column allow a field s type uniqueness nullability default column constraints etc to be modified oldfield is required to compute the necessary changes if strict is true raise errors if the old column does not match oldfield precisely ensure this field is even columnbased both sides have through models this is a noop always check constraints with the same mocked column name to avoid recreating constrains when the column is renamed perform a physical nonmanytomany field update drop any fk constraints we ll remake them later fksdropped set if self connection features supportsforeignkeys and oldfield remotefield and oldfield dbconstraint and self fieldshouldbealtered oldfield newfield ignoredbcomment fknames self constraintnames model oldfield column foreignkeytrue if strict and lenfknames 1 raise valueerror found wrong number s of foreign key constraints for s s lenfknames model meta dbtable oldfield column for fkname in fknames fksdropped addoldfield column self executeself deletefksqlmodel fkname has unique been removed if oldfield unique and not newfield unique or self fieldbecameprimarykeyoldfield newfield find the unique constraint for this field metaconstraintnames constraint name for constraint in model meta constraints constraintnames self constraintnames model oldfield column uniquetrue primarykeyfalse excludemetaconstraintnames if strict and lenconstraintnames 1 raise valueerror found wrong number s of unique constraints for s s lenconstraintnames model meta dbtable oldfield column for constraintname in constraintnames self executeself deleteuniquesqlmodel constraintname drop incoming fk constraints if the field is a primary key or unique which might be a tofield target and things are going to change oldcollation olddbparams getcollation newcollation newdbparams getcollation dropforeignkeys self connection features supportsforeignkeys and oldfield primarykey and newfield primarykey or oldfield unique and newfield unique and oldtype newtype or oldcollation newcollation if dropforeignkeys meta relatedfield also contains m2m reverse fields these will be filtered out for oldrel newrel in relatednonm2mobjectsoldfield newfield relfknames self constraintnames newrel relatedmodel newrel field column foreignkeytrue for fkname in relfknames self executeself deletefksqlnewrel relatedmodel fkname removed an index no strict check as multiple indexes are possible remove indexes if dbindex switched to false or a unique constraint will now be used in lieu of an index the following lines from the truth table show all true cases the rest are false oldfield dbindex oldfield unique newfield dbindex newfield unique true false false false true false false true true false true true if oldfield dbindex and not oldfield unique and not newfield dbindex or newfield unique find the index for this field metaindexnames index name for index in model meta indexes retrieve only btree indexes since this is what s created with dbindextrue indexnames self constraintnames model oldfield column indextrue typeindex suffix excludemetaindexnames for indexname in indexnames the only way to check if an index was created with dbindextrue or with index field name foo is to look at its name refs 28053 self executeself deleteindexsqlmodel indexname change check constraints olddbcheck self fielddbcheckoldfield olddbparams newdbcheck self fielddbchecknewfield newdbparams if olddbcheck newdbcheck and olddbcheck metaconstraintnames constraint name for constraint in model meta constraints constraintnames self constraintnames model oldfield column checktrue excludemetaconstraintnames if strict and lenconstraintnames 1 raise valueerror found wrong number s of check constraints for s s lenconstraintnames model meta dbtable oldfield column for constraintname in constraintnames self executeself deletechecksqlmodel constraintname have they renamed the column if oldfield column newfield column self execute self renamefieldsql model meta dbtable oldfield newfield newtype rename all references to the renamed column for sql in self deferredsql if isinstancesql statement sql renamecolumnreferences model meta dbtable oldfield column newfield column next start accumulating actions to do actions nullactions postactions type suffix change e g auto increment oldtypesuffix oldfield dbtypesuffixconnectionself connection newtypesuffix newfield dbtypesuffixconnectionself connection type collation or comment change if oldtype newtype or oldtypesuffix newtypesuffix or oldcollation newcollation or self connection features supportscomments and oldfield dbcomment newfield dbcomment fragment otheractions self altercolumntypesql model oldfield newfield newtype oldcollation newcollation actions appendfragment postactions extendotheractions if newfield dbdefault is not notprovided if oldfield dbdefault is notprovided or newfield dbdefault oldfield dbdefault actions append self altercolumndatabasedefaultsqlmodel oldfield newfield elif oldfield dbdefault is not notprovided actions append self altercolumndatabasedefaultsql model oldfield newfield droptrue when changing a column null constraint to not null with a given default value we need to perform 4 steps 1 add a default for new incoming writes 2 update existing null rows with new default 3 replace null constraint with not null 4 drop the default again default change needsdatabasedefault false if oldfield null and not newfield null and newfield dbdefault is notprovided olddefault self effectivedefaultoldfield newdefault self effectivedefaultnewfield if not self skipdefaultonalternewfield and olddefault newdefault and newdefault is not none needsdatabasedefault true actions append self altercolumndefaultsqlmodel oldfield newfield nullability change if oldfield null newfield null fragment self altercolumnnullsqlmodel oldfield newfield if fragment nullactions appendfragment only if we have a default and there is a change from null to not null fourwaydefaultalteration newfield hasdefault or newfield dbdefault is not notprovided and oldfield null and not newfield null if actions or nullactions if not fourwaydefaultalteration if we don t have to do a 4way default alteration we can directly run a not null alteration actions nullactions combine actions together if we can e g postgres if self connection features supportscombinedalters and actions sql params tuplezipactions actions joinsql sumparams apply those actions for sql params in actions self execute self sqlaltercolumn table self quotenamemodel meta dbtable changes sql params if fourwaydefaultalteration if newfield dbdefault is notprovided defaultsql s params newdefault else defaultsql params self dbdefaultsqlnewfield update existing rows with default value self execute self sqlupdatewithdefault table self quotenamemodel meta dbtable column self quotenamenewfield column default defaultsql params since we didn t run a not null change before we need to do it now for sql params in nullactions self execute self sqlaltercolumn table self quotenamemodel meta dbtable changes sql params if postactions for sql params in postactions self executesql params if primarykey changed to false delete the primary key constraint if oldfield primarykey and not newfield primarykey self deleteprimarykeymodel strict added a unique if self uniqueshouldbeaddedoldfield newfield self executeself createuniquesqlmodel newfield added an index add an index if dbindex switched to true or a unique constraint will no longer be used in lieu of an index the following lines from the truth table show all true cases the rest are false oldfield dbindex oldfield unique newfield dbindex newfield unique false false true false false true true false true true true false if not oldfield dbindex or oldfield unique and newfield dbindex and not newfield unique self executeself createindexsqlmodel fieldsnewfield type alteration on primary key then we need to alter the column referring to us relstoupdate if dropforeignkeys relstoupdate extendrelatednonm2mobjectsoldfield newfield changed to become primary key if self fieldbecameprimarykeyoldfield newfield make the new one self executeself createprimarykeysqlmodel newfield update all referencing columns relstoupdate extendrelatednonm2mobjectsoldfield newfield handle our type alters on the other end of rels from the pk stuff above for oldrel newrel in relstoupdate reldbparams newrel field dbparametersconnectionself connection reltype reldbparamstype relcollation reldbparams getcollation oldreldbparams oldrel field dbparametersconnectionself connection oldrelcollation oldreldbparams getcollation fragment otheractions self altercolumntypesql newrel relatedmodel oldrel field newrel field reltype oldrelcollation relcollation self execute self sqlaltercolumn table self quotenamenewrel relatedmodel meta dbtable changes fragment0 fragment1 for sql params in otheractions self executesql params does it have a foreign key if self connection features supportsforeignkeys and newfield remotefield and fksdropped or not oldfield remotefield or not oldfield dbconstraint and newfield dbconstraint self execute self createfksqlmodel newfield fktotablestocolumns rebuild fks that pointed to us if we previously had to drop them if dropforeignkeys for rel in relstoupdate if rel field dbconstraint self execute self createfksqlrel relatedmodel rel field fk does it have check constraints we need to add if olddbcheck newdbcheck and newdbcheck constraintname self createindexname model meta dbtable newfield column suffixcheck self execute self createchecksqlmodel constraintname newdbparamscheck drop the default if we need to django usually does not use indatabase defaults if needsdatabasedefault changessql params self altercolumndefaultsql model oldfield newfield droptrue sql self sqlaltercolumn table self quotenamemodel meta dbtable changes changessql self executesql params reset connection if required if self connection features connectionpersistsoldcolumns self connection close def altercolumnnullsqlself model oldfield newfield if self connection features interpretsemptystringsasnulls and newfield emptystringsallowed the field is nullable in the database anyway leave it alone return else newdbparams newfield dbparametersconnectionself connection sql self sqlaltercolumnnull if newfield null else self sqlaltercolumnnotnull return sql column self quotenamenewfield column type newdbparamstype def altercolumndefaultsqlself model oldfield newfield dropfalse newdefault self effectivedefaultnewfield default self columndefaultsqlnewfield params newdefault if drop params elif self connection features requiresliteraldefaults some databases oracle can t take defaults as a parameter if this is the case the schemaeditor for that database should implement preparedefault default self preparedefaultnewdefault params newdbparams newfield dbparametersconnectionself connection if drop if newfield null sql self sqlaltercolumnnodefaultnull else sql self sqlaltercolumnnodefault else sql self sqlaltercolumndefault return sql column self quotenamenewfield column type newdbparamstype default default params def altercolumndatabasedefaultsql self model oldfield newfield dropfalse if drop sql self sqlaltercolumnnodefault defaultsql params else sql self sqlaltercolumndefault defaultsql params self dbdefaultsqlnewfield newdbparams newfield dbparametersconnectionself connection return sql column self quotenamenewfield column type newdbparamstype default defaultsql params def altercolumntypesql self model oldfield newfield newtype oldcollation newcollation otheractions if collatesql self collatesql newcollation oldcollation model meta dbtable collatesql f collatesql else collatesql comment change commentsql if self connection features supportscomments and not newfield manytomany if oldfield dbcomment newfield dbcomment postgresql and oracle can t execute alter column and comment on at the same time sql params self altercolumncommentsql model newfield newtype newfield dbcomment if sql otheractions appendsql params if newfield dbcomment commentsql self commentsqlnewfield dbcomment return self sqlaltercolumntype column self quotenamenewfield column type newtype collation collatesql comment commentsql otheractions def altercolumncommentsqlself model newfield newtype newdbcomment return self sqlaltercolumncomment table self quotenamemodel meta dbtable column self quotenamenewfield column comment self commentsqlnewdbcomment def commentsqlself comment return self quotevaluecomment or def altermanytomanyself model oldfield newfield strict rename the through table repoint the fk to the other side the field that points to the target model is needed so we can tell alterfield to change it this is m2mreversefieldname as opposed to m2mfieldname which points to our model for selfreferential models we need to alter field from the other end too generate a unique name for an indexunique constraint the name is divided into 3 parts the table name the column names and a unique digest and suffix if everything fits into maxlength use that name shorten a long suffix prepend d if needed to prevent the name from starting with an underscore or a number not permitted on oracle return the sql statement to create the index for one or several fields or expressions sql can be specified if the syntax differs from the standard gis indexes return a list of all index sql statements field indexes meta indexes for the specified model return a list of all index sql statements for the specified field don t alter when changing only a field name changing an attribute that doesn t affect the schema changing an attribute in the provided set of ignored attributes adding only a dbcolumn and the column name is not changed dbtable does not change for model referenced by foreign keys databases support conditional covering functional unique and nulls distinct constraints via a unique index return all constraint names matching the columns and conditions if columnnames is not none columnnames self connection introspection identifierconverter truncatenamename self connection ops maxnamelength if self connection features truncatesnames else self connection introspection identifierconvertername for name in columnnames with self connection cursor as cursor constraints self connection introspection getconstraints cursor model meta dbtable result for name infodict in constraints items if columnnames is none or columnnames infodictcolumns if unique is not none and infodictunique unique continue if primarykey is not none and infodictprimarykey primarykey continue if index is not none and infodictindex index continue if check is not none and infodictcheck check continue if foreignkey is not none and not infodictforeignkey continue if type is not none and infodicttype type continue if not exclude or name not in exclude result appendname return result def deleteprimarykeyself model strictfalse constraintnames self constraintnamesmodel primarykeytrue if strict and lenconstraintnames 1 raise valueerror found wrong number s of pk constraints for s lenconstraintnames model meta dbtable for constraintname in constraintnames self executeself deleteprimarykeysqlmodel constraintname def createprimarykeysqlself model field return statement self sqlcreatepk tabletablemodel meta dbtable self quotename nameself quotename self createindexname model meta dbtable field column suffixpk columnscolumnsmodel meta dbtable field column self quotename def deleteprimarykeysqlself model name return self deleteconstraintsqlself sqldeletepk model name def collatesqlself collation oldcollationnone tablenamenone return collate self quotenamecollation if collation else def removeprocedureself procedurename paramtypes sql self sqldeleteprocedure procedure self quotenameprocedurename paramtypes joinparamtypes self executesql when altering the given field must constraints on its model from the given relation be temporarily dropped m2m reverse field foreign key constraint on the primary key which is being altered is the constraint targeting the field being altered related fields must be returned in a deterministic order filter out m2m objects from reverse relations return old_relation new_relation tuples this class and its subclasses are responsible for emitting schema changing statements to the databases model creation removal alteration field renaming index fiddling and so on overrideable sql templates state managing methods core utility functions execute the given sql statement with optional parameters don t perform the transactional ddl check if sql is being collected as it s not going to be executed anyway account for non string statement objects log the command we re running then run it take a model and return its table definition add any unique_togethers always deferred as some fields might be created afterward like geometry fields with some backends create column sql add fk deferreds if needed sql check constraints can go on the column sql here autoincrement sql for backends with inline variant fk add the sql to our big list autoincrement sql for backends with post table definition variant field database mapping functions work out nullability add database default include a default value if requested don t include a default value if it s a nullable field and the default cannot be dropped in the alter column statement e g mysql longtext and longblob some databases can t take defaults as a parameter oracle if this is the case the individual schema backend should implement prepare_default oracle treats the empty string as null so coerce the null option whenever is a possible value optionally add the tablespace if it s an implicitly indexed column return the column definition for a field the field must already have had set_attributes_from_name called get the column s type and use that as the basis of the sql check for fields that aren t actually columns e g m2m this appends to the params being returned some backends don t accept default values for certain columns types i e mysql longtext and longblob some backends don t accept default values for certain columns types i e mysql longtext and longblob in the alter column statement only used for backends which have requires_literal_defaults feature return the sql to use in a default clause the resulting string should contain a s placeholder for a default value return the sql and params for the field s database default some databases doesn t support parameterized defaults oracle sqlite if this is the case the individual schema backend should implement prepare_default return the sql to use in a generated always clause this method allows testing its logic without a connection return a field s effective database default value return a quoted version of the value so it s safe to use in an sql string this is not safe against injection from user code it is intended only for use in making sql scripts or preparing default values for particularly tricky backends defaults are not user defined though so this is safe actions create a table and any accompanying indexes or unique constraints for the given model prevent using as params in the case a literal is used in the definition add table comment add column comments add any field index deferred as sqlite _remake_table needs it make m2m tables delete a model from the database handle auto created intermediary models delete the table remove all deferred statements referencing the deleted table add an index on a model index create_sql returns interpolated sql which makes params none a necessity to avoid escaping attempts on execution remove an index from a model add a constraint to a model constraint create_sql returns interpolated sql which makes params none a necessity to avoid escaping attempts on execution remove a constraint from a model deal with a model changing its unique_together the input unique_togethers must be doubly nested not the single nested foo bar format deleted uniques created uniques deal with a model changing its index_together the input index_togethers must be doubly nested not the single nested foo bar format deleted indexes created indexes constraint matching the unique_together name rename the table a model points to rename all references to the old table name move a model s table between tablespaces create a field on a model usually involves adding a column but may involve adding a table instead for m2m fields special case implicit m2m tables get the column s definition it might not actually have a column behind it check constraints can go on the column sql here add fk constraint inline if supported otherwise add fk constraints later build the sql and run it drop the default if we need to add field comment if required add an index if required reset connection if required remove a field from a model usually involves deleting a column but for m2ms may involve deleting a table special case implicit m2m tables it might not actually have a column behind it drop any fk constraints mysql requires explicit deletion delete the column reset connection if required remove all deferred statements referencing the deleted column allow a field s type uniqueness nullability default column constraints etc to be modified old_field is required to compute the necessary changes if strict is true raise errors if the old column does not match old_field precisely ensure this field is even column based both sides have through models this is a no op always check constraints with the same mocked column name to avoid recreating constrains when the column is renamed perform a physical non manytomany field update drop any fk constraints we ll remake them later has unique been removed find the unique constraint for this field 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 _meta related_field also contains m2m reverse fields these will be filtered out 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 truth table show all true cases the rest are false old_field db_index old_field unique new_field db_index new_field unique true false false false true false false true true false true true find the index for this field retrieve only btree indexes since this is what s created with db_index true 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 change check constraints have they renamed the column rename all references to the renamed column next start accumulating actions to do type suffix change e g auto increment type collation or comment change when changing a column null constraint to not null with a given default value we need to perform 4 steps 1 add a default for new incoming writes 2 update existing null rows with new default 3 replace null constraint with not null 4 drop the default again default change nullability change only if we have a default and there is a change from null to not null if we don t have to do a 4 way default alteration we can directly run a not null alteration combine actions together if we can e g postgres apply those actions update existing rows with default value since we didn t run a not null change before we need to do it now if primary_key changed to false delete the primary key constraint added a unique added an index add an index if db_index switched to true or a unique constraint will no longer be used in lieu of an index the following lines from the truth table show all true cases the rest are false old_field db_index old_field unique new_field db_index new_field unique false false true false false true true false true true true false type alteration on primary key then we need to alter the column referring to us changed to become primary key make the new one update all referencing columns handle our type alters on the other end of rels from the pk stuff above does it have a foreign key rebuild fks that pointed to us if we previously had to drop them does it have check constraints we need to add drop the default if we need to django usually does not use in database defaults reset connection if required hook to specialize column null alteration return a sql params fragment to set a column to null or non null as required by new_field or none if no changes are required the field is nullable in the database anyway leave it alone hook to specialize column default alteration return a sql params fragment to add or drop depending on the drop argument a default to new_field s column some databases oracle can t take defaults as a parameter if this is the case the schemaeditor for that database should implement prepare_default hook to specialize column database default alteration return a sql params fragment to add or drop depending on the drop argument a default to new_field s column hook to specialize column type alteration for different backends for cases when a creation type is different to an alteration type e g serial in postgresql postgis fields return a 2 tuple of an sql fragment of sql params to insert into an alter table statement and a list of extra sql params tuples to run once the field is altered comment change postgresql and oracle can t execute alter column and comment on at the same time alter m2ms to repoint their to endpoints rename the through table repoint the fk to the other side the field that points to the target model is needed so we can tell alter_field to change it this is m2m_reverse_field_name as opposed to m2m_field_name which points to our model for self referential models we need to alter field from the other end too generate a unique name for an index unique constraint the name is divided into 3 parts the table name the column names and a unique digest and suffix if everything fits into max_length use that name shorten a long suffix prepend d if needed to prevent the name from starting with an underscore or a number not permitted on oracle return the sql statement to create the index for one or several fields or expressions sql can be specified if the syntax differs from the standard gis indexes return a list of all index sql statements field indexes meta indexes for the specified model return a list of all index sql statements for the specified field don t alter when changing only a field name changing an attribute that doesn t affect the schema changing an attribute in the provided set of ignored attributes adding only a db_column and the column name is not changed db_table does not change for model referenced by foreign keys databases support conditional covering functional unique and nulls distinct constraints via a unique index return all constraint names matching the columns and conditions
import logging
import operator
from datetime import datetime

from django.conf import settings
from django.db.backends.ddl_references import (
    Columns,
    Expressions,
    ForeignKeyName,
    IndexName,
    Statement,
    Table,
)
from django.db.backends.utils import names_digest, split_identifier, truncate_name
from django.db.models import NOT_PROVIDED, Deferrable, Index
from django.db.models.sql import Query
from django.db.transaction import TransactionManagementError, atomic
from django.utils import timezone

logger = logging.getLogger("django.db.backends.schema")


def _is_relevant_relation(relation, altered_field):
    
    field = relation.field
    if field.many_to_many:
        
        return False
    if altered_field.primary_key and field.to_fields == [None]:
        
        return True
    
    return altered_field.name in field.to_fields


def _all_related_fields(model):
    
    return sorted(
        model._meta._get_fields(
            forward=False,
            reverse=True,
            include_hidden=True,
            include_parents=False,
        ),
        key=operator.attrgetter("name"),
    )


def _related_non_m2m_objects(old_field, new_field):
    
    
    related_fields = zip(
        (
            obj
            for obj in _all_related_fields(old_field.model)
            if _is_relevant_relation(obj, old_field)
        ),
        (
            obj
            for obj in _all_related_fields(new_field.model)
            if _is_relevant_relation(obj, new_field)
        ),
    )
    for old_rel, new_rel in related_fields:
        yield old_rel, new_rel
        yield from _related_non_m2m_objects(
            old_rel.remote_field,
            new_rel.remote_field,
        )


class BaseDatabaseSchemaEditor:
    

    
    sql_create_table = "CREATE TABLE %(table)s (%(definition)s)"
    sql_rename_table = "ALTER TABLE %(old_table)s RENAME TO %(new_table)s"
    sql_retablespace_table = "ALTER TABLE %(table)s SET TABLESPACE %(new_tablespace)s"
    sql_delete_table = "DROP TABLE %(table)s CASCADE"

    sql_create_column = "ALTER TABLE %(table)s ADD COLUMN %(column)s %(definition)s"
    sql_alter_column = "ALTER TABLE %(table)s %(changes)s"
    sql_alter_column_type = "ALTER COLUMN %(column)s TYPE %(type)s%(collation)s"
    sql_alter_column_null = "ALTER COLUMN %(column)s DROP NOT NULL"
    sql_alter_column_not_null = "ALTER COLUMN %(column)s SET NOT NULL"
    sql_alter_column_default = "ALTER COLUMN %(column)s SET DEFAULT %(default)s"
    sql_alter_column_no_default = "ALTER COLUMN %(column)s DROP DEFAULT"
    sql_alter_column_no_default_null = sql_alter_column_no_default
    sql_delete_column = "ALTER TABLE %(table)s DROP COLUMN %(column)s CASCADE"
    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_unique_constraint = "UNIQUE (%(columns)s)%(deferrable)s"
    sql_check_constraint = "CHECK (%(check)s)"
    sql_delete_constraint = "ALTER TABLE %(table)s DROP CONSTRAINT %(name)s"
    sql_constraint = "CONSTRAINT %(name)s %(constraint)s"

    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%(nulls_distinct)s (%(columns)s)%(deferrable)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_create_column_inline_fk = None
    sql_delete_fk = sql_delete_constraint

    sql_create_index = (
        "CREATE INDEX %(name)s ON %(table)s "
        "(%(columns)s)%(include)s%(extra)s%(condition)s"
    )
    sql_create_unique_index = (
        "CREATE UNIQUE INDEX %(name)s ON %(table)s "
        "(%(columns)s)%(include)s%(condition)s%(nulls_distinct)s"
    )
    sql_rename_index = "ALTER INDEX %(old_name)s RENAME TO %(new_name)s"
    sql_delete_index = "DROP INDEX %(name)s"

    sql_create_pk = (
        "ALTER TABLE %(table)s ADD CONSTRAINT %(name)s PRIMARY KEY (%(columns)s)"
    )
    sql_delete_pk = sql_delete_constraint

    sql_delete_procedure = "DROP PROCEDURE %(procedure)s"

    sql_alter_table_comment = "COMMENT ON TABLE %(table)s IS %(comment)s"
    sql_alter_column_comment = "COMMENT ON COLUMN %(table)s.%(column)s IS %(comment)s"

    def __init__(self, connection, collect_sql=False, atomic=True):
        self.connection = connection
        self.collect_sql = collect_sql
        if self.collect_sql:
            self.collected_sql = []
        self.atomic_migration = self.connection.features.can_rollback_ddl and atomic

    

    def __enter__(self):
        self.deferred_sql = []
        if self.atomic_migration:
            self.atomic = atomic(self.connection.alias)
            self.atomic.__enter__()
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        if exc_type is None:
            for sql in self.deferred_sql:
                self.execute(sql)
        if self.atomic_migration:
            self.atomic.__exit__(exc_type, exc_value, traceback)

    

    def execute(self, sql, params=()):
        
        
        
        if (
            not self.collect_sql
            and self.connection.in_atomic_block
            and not self.connection.features.can_rollback_ddl
        ):
            raise TransactionManagementError(
                "Executing DDL statements while in a transaction on databases "
                "that can't perform a rollback is prohibited."
            )
        
        sql = str(sql)
        
        logger.debug(
            "%s; (params %r)", sql, params, extra={"params": params, "sql": sql}
        )
        if self.collect_sql:
            ending = "" if sql.rstrip().endswith(";") else ";"
            if params is not None:
                self.collected_sql.append(
                    (sql % tuple(map(self.quote_value, params))) + ending
                )
            else:
                self.collected_sql.append(sql + ending)
        else:
            with self.connection.cursor() as cursor:
                cursor.execute(sql, params)

    def quote_name(self, name):
        return self.connection.ops.quote_name(name)

    def table_sql(self, model):
        
        
        
        for field_names in model._meta.unique_together:
            fields = [model._meta.get_field(field) for field in field_names]
            self.deferred_sql.append(self._create_unique_sql(model, fields))
        
        column_sqls = []
        params = []
        for field in model._meta.local_fields:
            
            definition, extra_params = self.column_sql(model, field)
            if definition is None:
                continue
            
            db_params = field.db_parameters(connection=self.connection)
            if db_params["check"]:
                definition += " " + self.sql_check_constraint % db_params
            
            col_type_suffix = field.db_type_suffix(connection=self.connection)
            if col_type_suffix:
                definition += " %s" % col_type_suffix
            params.extend(extra_params)
            
            if field.remote_field and field.db_constraint:
                to_table = field.remote_field.model._meta.db_table
                to_column = field.remote_field.model._meta.get_field(
                    field.remote_field.field_name
                ).column
                if self.sql_create_inline_fk:
                    definition += " " + self.sql_create_inline_fk % {
                        "to_table": self.quote_name(to_table),
                        "to_column": self.quote_name(to_column),
                    }
                elif self.connection.features.supports_foreign_keys:
                    self.deferred_sql.append(
                        self._create_fk_sql(
                            model, field, "_fk_%(to_table)s_%(to_column)s"
                        )
                    )
            
            column_sqls.append(
                "%s %s"
                % (
                    self.quote_name(field.column),
                    definition,
                )
            )
            
            
            if field.get_internal_type() in (
                "AutoField",
                "BigAutoField",
                "SmallAutoField",
            ):
                autoinc_sql = self.connection.ops.autoinc_sql(
                    model._meta.db_table, field.column
                )
                if autoinc_sql:
                    self.deferred_sql.extend(autoinc_sql)
        constraints = [
            constraint.constraint_sql(model, self)
            for constraint in model._meta.constraints
        ]
        sql = self.sql_create_table % {
            "table": self.quote_name(model._meta.db_table),
            "definition": ", ".join(
                str(constraint)
                for constraint in (*column_sqls, *constraints)
                if constraint
            ),
        }
        if model._meta.db_tablespace:
            tablespace_sql = self.connection.ops.tablespace_sql(
                model._meta.db_tablespace
            )
            if tablespace_sql:
                sql += " " + tablespace_sql
        return sql, params

    

    def _iter_column_sql(
        self, column_db_type, params, model, field, field_db_params, include_default
    ):
        yield column_db_type
        if collation := field_db_params.get("collation"):
            yield self._collate_sql(collation)
        if self.connection.features.supports_comments_inline and field.db_comment:
            yield self._comment_sql(field.db_comment)
        
        null = field.null
        
        if field.db_default is not NOT_PROVIDED:
            default_sql, default_params = self.db_default_sql(field)
            yield f"DEFAULT {default_sql}"
            params.extend(default_params)
            include_default = False
        
        include_default = (
            include_default
            and not self.skip_default(field)
            and
            
            
            
            not (null and self.skip_default_on_alter(field))
        )
        if include_default:
            default_value = self.effective_default(field)
            if default_value is not None:
                column_default = "DEFAULT " + self._column_default_sql(field)
                if self.connection.features.requires_literal_defaults:
                    
                    
                    
                    yield column_default % self.prepare_default(default_value)
                else:
                    yield column_default
                    params.append(default_value)
        
        
        if (
            field.empty_strings_allowed
            and not field.primary_key
            and self.connection.features.interprets_empty_strings_as_nulls
        ):
            null = True
        if field.generated:
            yield self._column_generated_sql(field)
        elif not null:
            yield "NOT NULL"
        elif not self.connection.features.implied_column_null:
            yield "NULL"
        if field.primary_key:
            yield "PRIMARY KEY"
        elif field.unique:
            yield "UNIQUE"
        
        tablespace = field.db_tablespace or model._meta.db_tablespace
        if (
            tablespace
            and self.connection.features.supports_tablespaces
            and field.unique
        ):
            yield self.connection.ops.tablespace_sql(tablespace, inline=True)

    def column_sql(self, model, field, include_default=False):
        
        
        field_db_params = field.db_parameters(connection=self.connection)
        column_db_type = field_db_params["type"]
        
        if column_db_type is None:
            return None, None
        params = []
        return (
            " ".join(
                
                self._iter_column_sql(
                    column_db_type,
                    params,
                    model,
                    field,
                    field_db_params,
                    include_default,
                )
            ),
            params,
        )

    def skip_default(self, field):
        
        return False

    def skip_default_on_alter(self, field):
        
        return False

    def prepare_default(self, value):
        
        raise NotImplementedError(
            "subclasses of BaseDatabaseSchemaEditor for backends which have "
            "requires_literal_defaults must provide a prepare_default() method"
        )

    def _column_default_sql(self, field):
        
        return "%s"

    def db_default_sql(self, field):
        
        from django.db.models.expressions import Value

        sql = "%s" if isinstance(field.db_default, Value) else "(%s)"
        query = Query(model=field.model)
        compiler = query.get_compiler(connection=self.connection)
        default_sql, params = compiler.compile(field.db_default)
        if self.connection.features.requires_literal_defaults:
            
            
            
            default_sql %= tuple(self.prepare_default(p) for p in params)
            params = []
        return sql % default_sql, params

    def _column_generated_sql(self, field):
        
        expression_sql, params = field.generated_sql(self.connection)
        persistency_sql = "STORED" if field.db_persist else "VIRTUAL"
        if params:
            expression_sql = expression_sql % tuple(self.quote_value(p) for p in params)
        return f"GENERATED ALWAYS AS ({expression_sql}) {persistency_sql}"

    @staticmethod
    def _effective_default(field):
        
        if field.has_default():
            default = field.get_default()
        elif field.generated:
            default = None
        elif not field.null and field.blank and field.empty_strings_allowed:
            if field.get_internal_type() == "BinaryField":
                default = b""
            else:
                default = ""
        elif getattr(field, "auto_now", False) or getattr(field, "auto_now_add", False):
            internal_type = field.get_internal_type()
            if internal_type == "DateTimeField":
                default = timezone.now()
            else:
                default = datetime.now()
                if internal_type == "DateField":
                    default = default.date()
                elif internal_type == "TimeField":
                    default = default.time()
        else:
            default = None
        return default

    def effective_default(self, field):
        
        return field.get_db_prep_save(self._effective_default(field), self.connection)

    def quote_value(self, value):
        
        raise NotImplementedError()

    

    def create_model(self, model):
        
        sql, params = self.table_sql(model)
        
        
        self.execute(sql, params or None)

        if self.connection.features.supports_comments:
            
            if model._meta.db_table_comment:
                self.alter_db_table_comment(model, None, model._meta.db_table_comment)
            
            if not self.connection.features.supports_comments_inline:
                for field in model._meta.local_fields:
                    if field.db_comment:
                        field_db_params = field.db_parameters(
                            connection=self.connection
                        )
                        field_type = field_db_params["type"]
                        self.execute(
                            *self._alter_column_comment_sql(
                                model, field, field_type, field.db_comment
                            )
                        )
        
        self.deferred_sql.extend(self._model_indexes_sql(model))

        
        for field in model._meta.local_many_to_many:
            if field.remote_field.through._meta.auto_created:
                self.create_model(field.remote_field.through)

    def delete_model(self, model):
        
        
        for field in model._meta.local_many_to_many:
            if field.remote_field.through._meta.auto_created:
                self.delete_model(field.remote_field.through)

        
        self.execute(
            self.sql_delete_table
            % {
                "table": self.quote_name(model._meta.db_table),
            }
        )
        
        for sql in list(self.deferred_sql):
            if isinstance(sql, Statement) and sql.references_table(
                model._meta.db_table
            ):
                self.deferred_sql.remove(sql)

    def add_index(self, model, index):
        
        if (
            index.contains_expressions
            and not self.connection.features.supports_expression_indexes
        ):
            return None
        
        
        self.execute(index.create_sql(model, self), params=None)

    def remove_index(self, model, index):
        
        if (
            index.contains_expressions
            and not self.connection.features.supports_expression_indexes
        ):
            return None
        self.execute(index.remove_sql(model, self))

    def rename_index(self, model, old_index, new_index):
        if self.connection.features.can_rename_index:
            self.execute(
                self._rename_index_sql(model, old_index.name, new_index.name),
                params=None,
            )
        else:
            self.remove_index(model, old_index)
            self.add_index(model, new_index)

    def add_constraint(self, model, constraint):
        
        sql = constraint.create_sql(model, self)
        if sql:
            
            
            self.execute(sql, params=None)

    def remove_constraint(self, model, constraint):
        
        sql = constraint.remove_sql(model, self)
        if sql:
            self.execute(sql)

    def alter_unique_together(self, model, old_unique_together, new_unique_together):
        
        olds = {tuple(fields) for fields in old_unique_together}
        news = {tuple(fields) for fields in new_unique_together}
        
        for fields in olds.difference(news):
            self._delete_composed_index(
                model,
                fields,
                {"unique": True, "primary_key": False},
                self.sql_delete_unique,
            )
        
        for field_names in news.difference(olds):
            fields = [model._meta.get_field(field) for field in field_names]
            self.execute(self._create_unique_sql(model, fields))

    def alter_index_together(self, model, old_index_together, new_index_together):
        
        olds = {tuple(fields) for fields in old_index_together}
        news = {tuple(fields) for fields in new_index_together}
        
        for fields in olds.difference(news):
            self._delete_composed_index(
                model,
                fields,
                {"index": True, "unique": False},
                self.sql_delete_index,
            )
        
        for field_names in news.difference(olds):
            fields = [model._meta.get_field(field) for field in field_names]
            self.execute(self._create_index_sql(model, fields=fields, suffix="_idx"))

    def _delete_composed_index(self, model, fields, constraint_kwargs, sql):
        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]
        constraint_names = self._constraint_names(
            model,
            columns,
            exclude=meta_constraint_names | meta_index_names,
            **constraint_kwargs,
        )
        if (
            constraint_kwargs.get("unique") is True
            and constraint_names
            and self.connection.features.allows_multiple_constraints_on_same_fields
        ):
            
            default_name = str(
                self._unique_constraint_name(model._meta.db_table, columns, quote=False)
            )
            if default_name in constraint_names:
                constraint_names = [default_name]
        if len(constraint_names) != 1:
            raise ValueError(
                "Found wrong number (%s) of constraints for %s(%s)"
                % (
                    len(constraint_names),
                    model._meta.db_table,
                    ", ".join(columns),
                )
            )
        self.execute(self._delete_constraint_sql(sql, model, constraint_names[0]))

    def alter_db_table(self, model, old_db_table, new_db_table):
        
        if old_db_table == new_db_table or (
            self.connection.features.ignores_table_name_case
            and old_db_table.lower() == new_db_table.lower()
        ):
            return
        self.execute(
            self.sql_rename_table
            % {
                "old_table": self.quote_name(old_db_table),
                "new_table": self.quote_name(new_db_table),
            }
        )
        
        for sql in self.deferred_sql:
            if isinstance(sql, Statement):
                sql.rename_table_references(old_db_table, new_db_table)

    def alter_db_table_comment(self, model, old_db_table_comment, new_db_table_comment):
        if self.sql_alter_table_comment and self.connection.features.supports_comments:
            self.execute(
                self.sql_alter_table_comment
                % {
                    "table": self.quote_name(model._meta.db_table),
                    "comment": self.quote_value(new_db_table_comment or ""),
                }
            )

    def alter_db_tablespace(self, model, old_db_tablespace, new_db_tablespace):
        
        self.execute(
            self.sql_retablespace_table
            % {
                "table": self.quote_name(model._meta.db_table),
                "old_tablespace": self.quote_name(old_db_tablespace),
                "new_tablespace": self.quote_name(new_db_tablespace),
            }
        )

    def add_field(self, model, field):
        
        
        if field.many_to_many and field.remote_field.through._meta.auto_created:
            return self.create_model(field.remote_field.through)
        
        definition, params = self.column_sql(model, field, include_default=True)
        
        if definition is None:
            return
        if col_type_suffix := field.db_type_suffix(connection=self.connection):
            definition += f" {col_type_suffix}"
        
        db_params = field.db_parameters(connection=self.connection)
        if db_params["check"]:
            definition += " " + self.sql_check_constraint % db_params
        if (
            field.remote_field
            and self.connection.features.supports_foreign_keys
            and field.db_constraint
        ):
            constraint_suffix = "_fk_%(to_table)s_%(to_column)s"
            
            if self.sql_create_column_inline_fk:
                to_table = field.remote_field.model._meta.db_table
                to_column = field.remote_field.model._meta.get_field(
                    field.remote_field.field_name
                ).column
                namespace, _ = split_identifier(model._meta.db_table)
                definition += " " + self.sql_create_column_inline_fk % {
                    "name": self._fk_constraint_name(model, field, constraint_suffix),
                    "namespace": "%s." % self.quote_name(namespace)
                    if namespace
                    else "",
                    "column": self.quote_name(field.column),
                    "to_table": self.quote_name(to_table),
                    "to_column": self.quote_name(to_column),
                    "deferrable": self.connection.ops.deferrable_sql(),
                }
            
            else:
                self.deferred_sql.append(
                    self._create_fk_sql(model, field, constraint_suffix)
                )
        
        sql = self.sql_create_column % {
            "table": self.quote_name(model._meta.db_table),
            "column": self.quote_name(field.column),
            "definition": definition,
        }
        self.execute(sql, params)
        
        if (
            field.db_default is NOT_PROVIDED
            and not self.skip_default_on_alter(field)
            and self.effective_default(field) is not None
        ):
            changes_sql, params = self._alter_column_default_sql(
                model, None, field, drop=True
            )
            sql = self.sql_alter_column % {
                "table": self.quote_name(model._meta.db_table),
                "changes": changes_sql,
            }
            self.execute(sql, params)
        
        if (
            field.db_comment
            and self.connection.features.supports_comments
            and not self.connection.features.supports_comments_inline
        ):
            field_type = db_params["type"]
            self.execute(
                *self._alter_column_comment_sql(
                    model, field, field_type, field.db_comment
                )
            )
        
        self.deferred_sql.extend(self._field_indexes_sql(model, field))
        
        if self.connection.features.connection_persists_old_columns:
            self.connection.close()

    def remove_field(self, model, field):
        
        
        if field.many_to_many and field.remote_field.through._meta.auto_created:
            return self.delete_model(field.remote_field.through)
        
        if field.db_parameters(connection=self.connection)["type"] is None:
            return
        
        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_fk_sql(model, fk_name))
        
        sql = self.sql_delete_column % {
            "table": self.quote_name(model._meta.db_table),
            "column": self.quote_name(field.column),
        }
        self.execute(sql)
        
        if self.connection.features.connection_persists_old_columns:
            self.connection.close()
        
        for sql in list(self.deferred_sql):
            if isinstance(sql, Statement) and sql.references_column(
                model._meta.db_table, field.column
            ):
                self.deferred_sql.remove(sql)

    def alter_field(self, model, old_field, new_field, strict=False):
        
        if not self._field_should_be_altered(old_field, new_field):
            return
        
        old_db_params = old_field.db_parameters(connection=self.connection)
        old_type = old_db_params["type"]
        new_db_params = new_field.db_parameters(connection=self.connection)
        new_type = new_db_params["type"]
        if (old_type is None and old_field.remote_field is None) or (
            new_type is None and new_field.remote_field is None
        ):
            raise ValueError(
                "Cannot alter field %s into %s - they do not properly define "
                "db_type (are you using a badly-written custom field?)"
                % (old_field, new_field),
            )
        elif (
            old_type is None
            and new_type is None
            and (
                old_field.remote_field.through
                and new_field.remote_field.through
                and old_field.remote_field.through._meta.auto_created
                and new_field.remote_field.through._meta.auto_created
            )
        ):
            return self._alter_many_to_many(model, old_field, new_field, strict)
        elif (
            old_type is None
            and new_type is None
            and (
                old_field.remote_field.through
                and new_field.remote_field.through
                and not old_field.remote_field.through._meta.auto_created
                and not new_field.remote_field.through._meta.auto_created
            )
        ):
            
            return
        elif old_type is None or new_type is None:
            raise ValueError(
                "Cannot alter field %s into %s - they are not compatible types "
                "(you cannot alter to or from M2M fields, or add or remove "
                "through= on M2M fields)" % (old_field, new_field)
            )
        elif old_field.generated != new_field.generated or (
            new_field.generated
            and (
                old_field.db_persist != new_field.db_persist
                or old_field.generated_sql(self.connection)
                != new_field.generated_sql(self.connection)
            )
        ):
            raise ValueError(
                f"Modifying GeneratedFields is not supported - the field {new_field} "
                "must be removed and re-added with the new definition."
            )

        self._alter_field(
            model,
            old_field,
            new_field,
            old_type,
            new_type,
            old_db_params,
            new_db_params,
            strict,
        )

    def _field_db_check(self, field, field_db_params):
        
        
        check_constraints = self.connection.data_type_check_constraints
        data = field.db_type_parameters(self.connection)
        data["column"] = "__column_name__"
        try:
            return check_constraints[field.get_internal_type()] % data
        except KeyError:
            return None

    def _alter_field(
        self,
        model,
        old_field,
        new_field,
        old_type,
        new_type,
        old_db_params,
        new_db_params,
        strict=False,
    ):
        
        
        fks_dropped = set()
        if (
            self.connection.features.supports_foreign_keys
            and old_field.remote_field
            and old_field.db_constraint
            and self._field_should_be_altered(
                old_field,
                new_field,
                ignore={"db_comment"},
            )
        ):
            fk_names = self._constraint_names(
                model, [old_field.column], foreign_key=True
            )
            if strict and len(fk_names) != 1:
                raise ValueError(
                    "Found wrong number (%s) of foreign key constraints for %s.%s"
                    % (
                        len(fk_names),
                        model._meta.db_table,
                        old_field.column,
                    )
                )
            for fk_name in fk_names:
                fks_dropped.add((old_field.column,))
                self.execute(self._delete_fk_sql(model, fk_name))
        
        if old_field.unique and (
            not new_field.unique or self._field_became_primary_key(old_field, new_field)
        ):
            
            meta_constraint_names = {
                constraint.name for constraint in model._meta.constraints
            }
            constraint_names = self._constraint_names(
                model,
                [old_field.column],
                unique=True,
                primary_key=False,
                exclude=meta_constraint_names,
            )
            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_unique_sql(model, constraint_name))
        
        
        old_collation = old_db_params.get("collation")
        new_collation = new_db_params.get("collation")
        drop_foreign_keys = (
            self.connection.features.supports_foreign_keys
            and (
                (old_field.primary_key and new_field.primary_key)
                or (old_field.unique and new_field.unique)
            )
            and ((old_type != new_type) or (old_collation != new_collation))
        )
        if drop_foreign_keys:
            
            
            for _old_rel, new_rel in _related_non_m2m_objects(old_field, new_field):
                rel_fk_names = self._constraint_names(
                    new_rel.related_model, [new_rel.field.column], foreign_key=True
                )
                for fk_name in rel_fk_names:
                    self.execute(self._delete_fk_sql(new_rel.related_model, fk_name))
        
        
        
        
        
        
        
        
        
        
        if (
            old_field.db_index
            and not old_field.unique
            and (not new_field.db_index or new_field.unique)
        ):
            
            meta_index_names = {index.name for index in model._meta.indexes}
            
            
            index_names = self._constraint_names(
                model,
                [old_field.column],
                index=True,
                type_=Index.suffix,
                exclude=meta_index_names,
            )
            for index_name in index_names:
                
                
                
                self.execute(self._delete_index_sql(model, index_name))
        
        old_db_check = self._field_db_check(old_field, old_db_params)
        new_db_check = self._field_db_check(new_field, new_db_params)
        if old_db_check != new_db_check and old_db_check:
            meta_constraint_names = {
                constraint.name for constraint in model._meta.constraints
            }
            constraint_names = self._constraint_names(
                model,
                [old_field.column],
                check=True,
                exclude=meta_constraint_names,
            )
            if strict and len(constraint_names) != 1:
                raise ValueError(
                    "Found wrong number (%s) of check constraints for %s.%s"
                    % (
                        len(constraint_names),
                        model._meta.db_table,
                        old_field.column,
                    )
                )
            for constraint_name in constraint_names:
                self.execute(self._delete_check_sql(model, constraint_name))
        
        if old_field.column != new_field.column:
            self.execute(
                self._rename_field_sql(
                    model._meta.db_table, old_field, new_field, new_type
                )
            )
            
            for sql in self.deferred_sql:
                if isinstance(sql, Statement):
                    sql.rename_column_references(
                        model._meta.db_table, old_field.column, new_field.column
                    )
        
        actions = []
        null_actions = []
        post_actions = []
        
        old_type_suffix = old_field.db_type_suffix(connection=self.connection)
        new_type_suffix = new_field.db_type_suffix(connection=self.connection)
        
        if (
            old_type != new_type
            or old_type_suffix != new_type_suffix
            or old_collation != new_collation
            or (
                self.connection.features.supports_comments
                and old_field.db_comment != new_field.db_comment
            )
        ):
            fragment, other_actions = self._alter_column_type_sql(
                model, old_field, new_field, new_type, old_collation, new_collation
            )
            actions.append(fragment)
            post_actions.extend(other_actions)

        if new_field.db_default is not NOT_PROVIDED:
            if (
                old_field.db_default is NOT_PROVIDED
                or new_field.db_default != old_field.db_default
            ):
                actions.append(
                    self._alter_column_database_default_sql(model, old_field, new_field)
                )
        elif old_field.db_default is not NOT_PROVIDED:
            actions.append(
                self._alter_column_database_default_sql(
                    model, old_field, new_field, drop=True
                )
            )
        
        
        
        
        
        
        
        needs_database_default = False
        if (
            old_field.null
            and not new_field.null
            and new_field.db_default is NOT_PROVIDED
        ):
            old_default = self.effective_default(old_field)
            new_default = self.effective_default(new_field)
            if (
                not self.skip_default_on_alter(new_field)
                and old_default != new_default
                and new_default is not None
            ):
                needs_database_default = True
                actions.append(
                    self._alter_column_default_sql(model, old_field, new_field)
                )
        
        if old_field.null != new_field.null:
            fragment = self._alter_column_null_sql(model, old_field, new_field)
            if fragment:
                null_actions.append(fragment)
        
        four_way_default_alteration = (
            new_field.has_default() or new_field.db_default is not NOT_PROVIDED
        ) and (old_field.null and not new_field.null)
        if actions or null_actions:
            if not four_way_default_alteration:
                
                
                actions += null_actions
            
            if self.connection.features.supports_combined_alters and actions:
                sql, params = tuple(zip(*actions))
                actions = [(", ".join(sql), sum(params, []))]
            
            for sql, params in actions:
                self.execute(
                    self.sql_alter_column
                    % {
                        "table": self.quote_name(model._meta.db_table),
                        "changes": sql,
                    },
                    params,
                )
            if four_way_default_alteration:
                if new_field.db_default is NOT_PROVIDED:
                    default_sql = "%s"
                    params = [new_default]
                else:
                    default_sql, params = self.db_default_sql(new_field)
                
                self.execute(
                    self.sql_update_with_default
                    % {
                        "table": self.quote_name(model._meta.db_table),
                        "column": self.quote_name(new_field.column),
                        "default": default_sql,
                    },
                    params,
                )
                
                
                for sql, params in null_actions:
                    self.execute(
                        self.sql_alter_column
                        % {
                            "table": self.quote_name(model._meta.db_table),
                            "changes": sql,
                        },
                        params,
                    )
        if post_actions:
            for sql, params in post_actions:
                self.execute(sql, params)
        
        if old_field.primary_key and not new_field.primary_key:
            self._delete_primary_key(model, strict)
        
        if self._unique_should_be_added(old_field, new_field):
            self.execute(self._create_unique_sql(model, [new_field]))
        
        
        
        
        
        
        
        
        
        if (
            (not old_field.db_index or old_field.unique)
            and new_field.db_index
            and not new_field.unique
        ):
            self.execute(self._create_index_sql(model, fields=[new_field]))
        
        
        rels_to_update = []
        if drop_foreign_keys:
            rels_to_update.extend(_related_non_m2m_objects(old_field, new_field))
        
        if self._field_became_primary_key(old_field, new_field):
            
            self.execute(self._create_primary_key_sql(model, new_field))
            
            rels_to_update.extend(_related_non_m2m_objects(old_field, new_field))
        
        for old_rel, new_rel in rels_to_update:
            rel_db_params = new_rel.field.db_parameters(connection=self.connection)
            rel_type = rel_db_params["type"]
            rel_collation = rel_db_params.get("collation")
            old_rel_db_params = old_rel.field.db_parameters(connection=self.connection)
            old_rel_collation = old_rel_db_params.get("collation")
            fragment, other_actions = self._alter_column_type_sql(
                new_rel.related_model,
                old_rel.field,
                new_rel.field,
                rel_type,
                old_rel_collation,
                rel_collation,
            )
            self.execute(
                self.sql_alter_column
                % {
                    "table": self.quote_name(new_rel.related_model._meta.db_table),
                    "changes": fragment[0],
                },
                fragment[1],
            )
            for sql, params in other_actions:
                self.execute(sql, params)
        
        if (
            self.connection.features.supports_foreign_keys
            and new_field.remote_field
            and (
                fks_dropped or not old_field.remote_field or not old_field.db_constraint
            )
            and new_field.db_constraint
        ):
            self.execute(
                self._create_fk_sql(model, new_field, "_fk_%(to_table)s_%(to_column)s")
            )
        
        if drop_foreign_keys:
            for _, rel in rels_to_update:
                if rel.field.db_constraint:
                    self.execute(
                        self._create_fk_sql(rel.related_model, rel.field, "_fk")
                    )
        
        if old_db_check != new_db_check and new_db_check:
            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"])
            )
        
        
        if needs_database_default:
            changes_sql, params = self._alter_column_default_sql(
                model, old_field, new_field, drop=True
            )
            sql = self.sql_alter_column % {
                "table": self.quote_name(model._meta.db_table),
                "changes": changes_sql,
            }
            self.execute(sql, params)
        
        if self.connection.features.connection_persists_old_columns:
            self.connection.close()

    def _alter_column_null_sql(self, model, old_field, new_field):
        
        if (
            self.connection.features.interprets_empty_strings_as_nulls
            and new_field.empty_strings_allowed
        ):
            
            return
        else:
            new_db_params = new_field.db_parameters(connection=self.connection)
            sql = (
                self.sql_alter_column_null
                if new_field.null
                else self.sql_alter_column_not_null
            )
            return (
                sql
                % {
                    "column": self.quote_name(new_field.column),
                    "type": new_db_params["type"],
                },
                [],
            )

    def _alter_column_default_sql(self, model, old_field, new_field, drop=False):
        
        new_default = self.effective_default(new_field)
        default = self._column_default_sql(new_field)
        params = [new_default]

        if drop:
            params = []
        elif self.connection.features.requires_literal_defaults:
            
            
            
            default = self.prepare_default(new_default)
            params = []

        new_db_params = new_field.db_parameters(connection=self.connection)
        if drop:
            if new_field.null:
                sql = self.sql_alter_column_no_default_null
            else:
                sql = self.sql_alter_column_no_default
        else:
            sql = self.sql_alter_column_default
        return (
            sql
            % {
                "column": self.quote_name(new_field.column),
                "type": new_db_params["type"],
                "default": default,
            },
            params,
        )

    def _alter_column_database_default_sql(
        self, model, old_field, new_field, drop=False
    ):
        
        if drop:
            sql = self.sql_alter_column_no_default
            default_sql = ""
            params = []
        else:
            sql = self.sql_alter_column_default
            default_sql, params = self.db_default_sql(new_field)

        new_db_params = new_field.db_parameters(connection=self.connection)
        return (
            sql
            % {
                "column": self.quote_name(new_field.column),
                "type": new_db_params["type"],
                "default": default_sql,
            },
            params,
        )

    def _alter_column_type_sql(
        self, model, old_field, new_field, new_type, old_collation, new_collation
    ):
        
        other_actions = []
        if collate_sql := self._collate_sql(
            new_collation, old_collation, model._meta.db_table
        ):
            collate_sql = f" {collate_sql}"
        else:
            collate_sql = ""
        
        comment_sql = ""
        if self.connection.features.supports_comments and not new_field.many_to_many:
            if old_field.db_comment != new_field.db_comment:
                
                
                sql, params = self._alter_column_comment_sql(
                    model, new_field, new_type, new_field.db_comment
                )
                if sql:
                    other_actions.append((sql, params))
            if new_field.db_comment:
                comment_sql = self._comment_sql(new_field.db_comment)
        return (
            (
                self.sql_alter_column_type
                % {
                    "column": self.quote_name(new_field.column),
                    "type": new_type,
                    "collation": collate_sql,
                    "comment": comment_sql,
                },
                [],
            ),
            other_actions,
        )

    def _alter_column_comment_sql(self, model, new_field, new_type, new_db_comment):
        return (
            self.sql_alter_column_comment
            % {
                "table": self.quote_name(model._meta.db_table),
                "column": self.quote_name(new_field.column),
                "comment": self._comment_sql(new_db_comment),
            },
            [],
        )

    def _comment_sql(self, comment):
        return self.quote_value(comment or "")

    def _alter_many_to_many(self, model, old_field, new_field, strict):
        
        
        if (
            old_field.remote_field.through._meta.db_table
            != new_field.remote_field.through._meta.db_table
        ):
            self.alter_db_table(
                old_field.remote_field.through,
                old_field.remote_field.through._meta.db_table,
                new_field.remote_field.through._meta.db_table,
            )
        
        self.alter_field(
            new_field.remote_field.through,
            
            
            
            old_field.remote_field.through._meta.get_field(
                old_field.m2m_reverse_field_name()
            ),
            new_field.remote_field.through._meta.get_field(
                new_field.m2m_reverse_field_name()
            ),
        )
        self.alter_field(
            new_field.remote_field.through,
            
            old_field.remote_field.through._meta.get_field(old_field.m2m_field_name()),
            new_field.remote_field.through._meta.get_field(new_field.m2m_field_name()),
        )

    def _create_index_name(self, table_name, column_names, suffix=""):
        
        _, table_name = split_identifier(table_name)
        hash_suffix_part = "%s%s" % (
            names_digest(table_name, *column_names, length=8),
            suffix,
        )
        max_length = self.connection.ops.max_name_length() or 200
        
        index_name = "%s_%s_%s" % (table_name, "_".join(column_names), hash_suffix_part)
        if len(index_name) <= max_length:
            return index_name
        
        if len(hash_suffix_part) > max_length / 3:
            hash_suffix_part = hash_suffix_part[: max_length // 3]
        other_length = (max_length - len(hash_suffix_part)) // 2 - 1
        index_name = "%s_%s_%s" % (
            table_name[:other_length],
            "_".join(column_names)[:other_length],
            hash_suffix_part,
        )
        
        
        if index_name[0] == "_" or index_name[0].isdigit():
            index_name = "D%s" % index_name[:-1]
        return index_name

    def _get_index_tablespace_sql(self, model, fields, db_tablespace=None):
        if db_tablespace is None:
            if len(fields) == 1 and fields[0].db_tablespace:
                db_tablespace = fields[0].db_tablespace
            elif settings.DEFAULT_INDEX_TABLESPACE:
                db_tablespace = settings.DEFAULT_INDEX_TABLESPACE
            elif model._meta.db_tablespace:
                db_tablespace = model._meta.db_tablespace
        if db_tablespace is not None:
            return " " + self.connection.ops.tablespace_sql(db_tablespace)
        return ""

    def _index_condition_sql(self, condition):
        if condition:
            return " WHERE " + condition
        return ""

    def _index_include_sql(self, model, columns):
        if not columns or not self.connection.features.supports_covering_indexes:
            return ""
        return Statement(
            " INCLUDE (%(columns)s)",
            columns=Columns(model._meta.db_table, columns, self.quote_name),
        )

    def _create_index_sql(
        self,
        model,
        *,
        fields=None,
        name=None,
        suffix="",
        using="",
        db_tablespace=None,
        col_suffixes=(),
        sql=None,
        opclasses=(),
        condition=None,
        include=None,
        expressions=None,
    ):
        
        fields = fields or []
        expressions = expressions or []
        compiler = Query(model, alias_cols=False).get_compiler(
            connection=self.connection,
        )
        tablespace_sql = self._get_index_tablespace_sql(
            model, fields, db_tablespace=db_tablespace
        )
        columns = [field.column for field in fields]
        sql_create_index = sql or self.sql_create_index
        table = model._meta.db_table

        def create_index_name(*args, **kwargs):
            nonlocal name
            if name is None:
                name = self._create_index_name(*args, **kwargs)
            return self.quote_name(name)

        return Statement(
            sql_create_index,
            table=Table(table, self.quote_name),
            name=IndexName(table, columns, suffix, create_index_name),
            using=using,
            columns=(
                self._index_columns(table, columns, col_suffixes, opclasses)
                if columns
                else Expressions(table, expressions, compiler, self.quote_value)
            ),
            extra=tablespace_sql,
            condition=self._index_condition_sql(condition),
            include=self._index_include_sql(model, include),
        )

    def _delete_index_sql(self, model, name, sql=None):
        return Statement(
            sql or self.sql_delete_index,
            table=Table(model._meta.db_table, self.quote_name),
            name=self.quote_name(name),
        )

    def _rename_index_sql(self, model, old_name, new_name):
        return Statement(
            self.sql_rename_index,
            table=Table(model._meta.db_table, self.quote_name),
            old_name=self.quote_name(old_name),
            new_name=self.quote_name(new_name),
        )

    def _index_columns(self, table, columns, col_suffixes, opclasses):
        return Columns(table, columns, self.quote_name, col_suffixes=col_suffixes)

    def _model_indexes_sql(self, 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 index in model._meta.indexes:
            if (
                not index.contains_expressions
                or self.connection.features.supports_expression_indexes
            ):
                output.append(index.create_sql(model, self))
        return output

    def _field_indexes_sql(self, model, field):
        
        output = []
        if self._field_should_be_indexed(model, field):
            output.append(self._create_index_sql(model, fields=[field]))
        return output

    def _field_should_be_altered(self, old_field, new_field, ignore=None):
        if not old_field.concrete and not new_field.concrete:
            return False
        ignore = ignore or set()
        _, old_path, old_args, old_kwargs = old_field.deconstruct()
        _, new_path, new_args, new_kwargs = new_field.deconstruct()
        
        
        
        
        
        
        for attr in ignore.union(old_field.non_db_attrs):
            old_kwargs.pop(attr, None)
        for attr in ignore.union(new_field.non_db_attrs):
            new_kwargs.pop(attr, None)
        if (
            not new_field.many_to_many
            and old_field.remote_field
            and new_field.remote_field
            and old_field.remote_field.model._meta.db_table
            == new_field.remote_field.model._meta.db_table
        ):
            old_kwargs.pop("to", None)
            new_kwargs.pop("to", None)
        return self.quote_name(old_field.column) != self.quote_name(
            new_field.column
        ) or (old_path, old_args, old_kwargs) != (new_path, new_args, new_kwargs)

    def _field_should_be_indexed(self, model, field):
        return field.db_index and not field.unique

    def _field_became_primary_key(self, old_field, new_field):
        return not old_field.primary_key and new_field.primary_key

    def _unique_should_be_added(self, old_field, new_field):
        return (
            not new_field.primary_key
            and new_field.unique
            and (not old_field.unique or old_field.primary_key)
        )

    def _rename_field_sql(self, table, old_field, new_field, new_type):
        return self.sql_rename_column % {
            "table": self.quote_name(table),
            "old_column": self.quote_name(old_field.column),
            "new_column": self.quote_name(new_field.column),
            "type": new_type,
        }

    def _create_fk_sql(self, model, field, suffix):
        table = Table(model._meta.db_table, self.quote_name)
        name = self._fk_constraint_name(model, field, suffix)
        column = Columns(model._meta.db_table, [field.column], self.quote_name)
        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()
        return Statement(
            self.sql_create_fk,
            table=table,
            name=name,
            column=column,
            to_table=to_table,
            to_column=to_column,
            deferrable=deferrable,
        )

    def _fk_constraint_name(self, model, field, suffix):
        def create_fk_name(*args, **kwargs):
            return self.quote_name(self._create_index_name(*args, **kwargs))

        return ForeignKeyName(
            model._meta.db_table,
            [field.column],
            split_identifier(field.target_field.model._meta.db_table)[1],
            [field.target_field.column],
            suffix,
            create_fk_name,
        )

    def _delete_fk_sql(self, model, name):
        return self._delete_constraint_sql(self.sql_delete_fk, model, name)

    def _deferrable_constraint_sql(self, deferrable):
        if deferrable is None:
            return ""
        if deferrable == Deferrable.DEFERRED:
            return " DEFERRABLE INITIALLY DEFERRED"
        if deferrable == Deferrable.IMMEDIATE:
            return " DEFERRABLE INITIALLY IMMEDIATE"

    def _unique_index_nulls_distinct_sql(self, nulls_distinct):
        if nulls_distinct is False:
            return " NULLS NOT DISTINCT"
        elif nulls_distinct is True:
            return " NULLS DISTINCT"
        return ""

    def _unique_supported(
        self,
        condition=None,
        deferrable=None,
        include=None,
        expressions=None,
        nulls_distinct=None,
    ):
        return (
            (not condition or self.connection.features.supports_partial_indexes)
            and (
                not deferrable
                or self.connection.features.supports_deferrable_unique_constraints
            )
            and (not include or self.connection.features.supports_covering_indexes)
            and (
                not expressions or self.connection.features.supports_expression_indexes
            )
            and (
                nulls_distinct is None
                or self.connection.features.supports_nulls_distinct_unique_constraints
            )
        )

    def _unique_sql(
        self,
        model,
        fields,
        name,
        condition=None,
        deferrable=None,
        include=None,
        opclasses=None,
        expressions=None,
        nulls_distinct=None,
    ):
        if not self._unique_supported(
            condition=condition,
            deferrable=deferrable,
            include=include,
            expressions=expressions,
            nulls_distinct=nulls_distinct,
        ):
            return None

        if (
            condition
            or include
            or opclasses
            or expressions
            or nulls_distinct is not None
        ):
            
            
            sql = self._create_unique_sql(
                model,
                fields,
                name=name,
                condition=condition,
                include=include,
                opclasses=opclasses,
                expressions=expressions,
                nulls_distinct=nulls_distinct,
            )
            if sql:
                self.deferred_sql.append(sql)
            return None
        constraint = self.sql_unique_constraint % {
            "columns": ", ".join([self.quote_name(field.column) for field in fields]),
            "deferrable": self._deferrable_constraint_sql(deferrable),
        }
        return self.sql_constraint % {
            "name": self.quote_name(name),
            "constraint": constraint,
        }

    def _create_unique_sql(
        self,
        model,
        fields,
        name=None,
        condition=None,
        deferrable=None,
        include=None,
        opclasses=None,
        expressions=None,
        nulls_distinct=None,
    ):
        if not self._unique_supported(
            condition=condition,
            deferrable=deferrable,
            include=include,
            expressions=expressions,
            nulls_distinct=nulls_distinct,
        ):
            return None

        compiler = Query(model, alias_cols=False).get_compiler(
            connection=self.connection
        )
        table = model._meta.db_table
        columns = [field.column for field in fields]
        if name is None:
            name = self._unique_constraint_name(table, columns, quote=True)
        else:
            name = self.quote_name(name)
        if condition or include or opclasses or expressions:
            sql = self.sql_create_unique_index
        else:
            sql = self.sql_create_unique
        if columns:
            columns = self._index_columns(
                table, columns, col_suffixes=(), opclasses=opclasses
            )
        else:
            columns = Expressions(table, expressions, compiler, self.quote_value)
        return Statement(
            sql,
            table=Table(table, self.quote_name),
            name=name,
            columns=columns,
            condition=self._index_condition_sql(condition),
            deferrable=self._deferrable_constraint_sql(deferrable),
            include=self._index_include_sql(model, include),
            nulls_distinct=self._unique_index_nulls_distinct_sql(nulls_distinct),
        )

    def _unique_constraint_name(self, table, columns, quote=True):
        if quote:

            def create_unique_name(*args, **kwargs):
                return self.quote_name(self._create_index_name(*args, **kwargs))

        else:
            create_unique_name = self._create_index_name

        return IndexName(table, columns, "_uniq", create_unique_name)

    def _delete_unique_sql(
        self,
        model,
        name,
        condition=None,
        deferrable=None,
        include=None,
        opclasses=None,
        expressions=None,
        nulls_distinct=None,
    ):
        if not self._unique_supported(
            condition=condition,
            deferrable=deferrable,
            include=include,
            expressions=expressions,
            nulls_distinct=nulls_distinct,
        ):
            return None
        if condition or include or opclasses or expressions:
            sql = self.sql_delete_index
        else:
            sql = self.sql_delete_unique
        return self._delete_constraint_sql(sql, 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):
        if not self.connection.features.supports_table_check_constraints:
            return None
        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):
        if not self.connection.features.supports_table_check_constraints:
            return None
        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,
        exclude=None,
    ):
        
        if column_names is not None:
            column_names = [
                self.connection.introspection.identifier_converter(
                    truncate_name(name, self.connection.ops.max_name_length())
                )
                if self.connection.features.truncates_names
                else self.connection.introspection.identifier_converter(name)
                for name in column_names
            ]
        with self.connection.cursor() as cursor:
            constraints = self.connection.introspection.get_constraints(
                cursor, model._meta.db_table
            )
        result = []
        for name, infodict in constraints.items():
            if column_names is None or column_names == infodict["columns"]:
                if unique is not None and infodict["unique"] != unique:
                    continue
                if primary_key is not None and infodict["primary_key"] != primary_key:
                    continue
                if index is not None and infodict["index"] != index:
                    continue
                if check is not None and infodict["check"] != check:
                    continue
                if foreign_key is not None and not infodict["foreign_key"]:
                    continue
                if type_ is not None and infodict["type"] != type_:
                    continue
                if not exclude or name not in exclude:
                    result.append(name)
        return result

    def _delete_primary_key(self, model, strict=False):
        constraint_names = self._constraint_names(model, primary_key=True)
        if strict and len(constraint_names) != 1:
            raise ValueError(
                "Found wrong number (%s) of PK constraints for %s"
                % (
                    len(constraint_names),
                    model._meta.db_table,
                )
            )
        for constraint_name in constraint_names:
            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 _collate_sql(self, collation, old_collation=None, table_name=None):
        return "COLLATE " + self.quote_name(collation) if collation else ""

    def remove_procedure(self, procedure_name, param_types=()):
        sql = self.sql_delete_procedure % {
            "procedure": self.quote_name(procedure_name),
            "param_types": ",".join(param_types),
        }
        self.execute(sql)