1830 lines
72 KiB
Python
1830 lines
72 KiB
Python
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 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):
|
|
"""
|
|
When altering the given field, must constraints on its model from the given
|
|
relation be temporarily dropped?
|
|
"""
|
|
field = relation.field
|
|
if field.many_to_many:
|
|
# M2M reverse field
|
|
return False
|
|
if altered_field.primary_key and field.to_fields == [None]:
|
|
# Foreign key constraint on the primary key, which is being altered.
|
|
return True
|
|
# Is the constraint targeting the field being altered?
|
|
return altered_field.name in field.to_fields
|
|
|
|
|
|
def _all_related_fields(model):
|
|
# Related fields must be returned in a deterministic order.
|
|
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):
|
|
# Filter out m2m objects from reverse relations.
|
|
# Return (old_relation, new_relation) tuples.
|
|
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:
|
|
"""
|
|
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
|
|
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 (%(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"
|
|
)
|
|
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
|
|
|
|
# State-managing methods
|
|
|
|
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)
|
|
|
|
# Core utility functions
|
|
|
|
def execute(self, sql, params=()):
|
|
"""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.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."
|
|
)
|
|
# Account for non-string statement objects.
|
|
sql = str(sql)
|
|
# Log the command we're running, then run it
|
|
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):
|
|
"""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).
|
|
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))
|
|
# Create column SQL, add FK deferreds if needed.
|
|
column_sqls = []
|
|
params = []
|
|
for field in model._meta.local_fields:
|
|
# SQL.
|
|
definition, extra_params = self.column_sql(model, field)
|
|
if definition is None:
|
|
continue
|
|
# Check constraints can go on the column SQL here.
|
|
db_params = field.db_parameters(connection=self.connection)
|
|
if db_params["check"]:
|
|
definition += " " + self.sql_check_constraint % db_params
|
|
# Autoincrement SQL (for backends with inline variant).
|
|
col_type_suffix = field.db_type_suffix(connection=self.connection)
|
|
if col_type_suffix:
|
|
definition += " %s" % col_type_suffix
|
|
params.extend(extra_params)
|
|
# FK.
|
|
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"
|
|
)
|
|
)
|
|
# Add the SQL to our big list.
|
|
column_sqls.append(
|
|
"%s %s"
|
|
% (
|
|
self.quote_name(field.column),
|
|
definition,
|
|
)
|
|
)
|
|
# Autoincrement SQL (for backends with post table definition
|
|
# variant).
|
|
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
|
|
|
|
# Field <-> database mapping functions
|
|
|
|
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)
|
|
# Work out nullability.
|
|
null = field.null
|
|
# Include a default value, if requested.
|
|
include_default = (
|
|
include_default
|
|
and not self.skip_default(field)
|
|
and
|
|
# 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).
|
|
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:
|
|
# Some databases can't take defaults as a parameter (Oracle).
|
|
# If this is the case, the individual schema backend should
|
|
# implement prepare_default().
|
|
yield column_default % self.prepare_default(default_value)
|
|
else:
|
|
yield column_default
|
|
params.append(default_value)
|
|
# Oracle treats the empty string ('') as null, so coerce the null
|
|
# option whenever '' is a possible value.
|
|
if (
|
|
field.empty_strings_allowed
|
|
and not field.primary_key
|
|
and self.connection.features.interprets_empty_strings_as_nulls
|
|
):
|
|
null = True
|
|
if 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"
|
|
# Optionally add the tablespace if it's an implicitly indexed column.
|
|
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):
|
|
"""
|
|
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.
|
|
field_db_params = field.db_parameters(connection=self.connection)
|
|
column_db_type = field_db_params["type"]
|
|
# Check for fields that aren't actually columns (e.g. M2M).
|
|
if column_db_type is None:
|
|
return None, None
|
|
params = []
|
|
return (
|
|
" ".join(
|
|
# This appends to the params being returned.
|
|
self._iter_column_sql(
|
|
column_db_type,
|
|
params,
|
|
model,
|
|
field,
|
|
field_db_params,
|
|
include_default,
|
|
)
|
|
),
|
|
params,
|
|
)
|
|
|
|
def skip_default(self, field):
|
|
"""
|
|
Some backends don't accept default values for certain columns types
|
|
(i.e. MySQL longtext and longblob).
|
|
"""
|
|
return False
|
|
|
|
def skip_default_on_alter(self, field):
|
|
"""
|
|
Some backends don't accept default values for certain columns types
|
|
(i.e. MySQL longtext and longblob) in the ALTER COLUMN statement.
|
|
"""
|
|
return False
|
|
|
|
def prepare_default(self, value):
|
|
"""
|
|
Only used for backends which have requires_literal_defaults feature
|
|
"""
|
|
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 the SQL to use in a DEFAULT clause. The resulting string should
|
|
contain a '%s' placeholder for a default value.
|
|
"""
|
|
return "%s"
|
|
|
|
@staticmethod
|
|
def _effective_default(field):
|
|
# This method allows testing its logic without a connection.
|
|
if field.has_default():
|
|
default = field.get_default()
|
|
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 a field's effective database default value."""
|
|
return field.get_db_prep_save(self._effective_default(field), self.connection)
|
|
|
|
def quote_value(self, 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).
|
|
"""
|
|
raise NotImplementedError()
|
|
|
|
# Actions
|
|
|
|
def create_model(self, model):
|
|
"""
|
|
Create a table and any accompanying indexes or unique constraints for
|
|
the given `model`.
|
|
"""
|
|
sql, params = self.table_sql(model)
|
|
# Prevent using [] as params, in the case a literal '%' is used in the
|
|
# definition.
|
|
self.execute(sql, params or None)
|
|
|
|
if self.connection.features.supports_comments:
|
|
# Add table comment.
|
|
if model._meta.db_table_comment:
|
|
self.alter_db_table_comment(model, None, model._meta.db_table_comment)
|
|
# Add column comments.
|
|
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
|
|
)
|
|
)
|
|
# Add any field index and index_together's (deferred as SQLite
|
|
# _remake_table needs it).
|
|
self.deferred_sql.extend(self._model_indexes_sql(model))
|
|
|
|
# Make M2M tables
|
|
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):
|
|
"""Delete a model from the database."""
|
|
# Handle auto-created intermediary models
|
|
for field in model._meta.local_many_to_many:
|
|
if field.remote_field.through._meta.auto_created:
|
|
self.delete_model(field.remote_field.through)
|
|
|
|
# Delete the table
|
|
self.execute(
|
|
self.sql_delete_table
|
|
% {
|
|
"table": self.quote_name(model._meta.db_table),
|
|
}
|
|
)
|
|
# Remove all deferred statements referencing the deleted 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):
|
|
"""Add an index on a model."""
|
|
if (
|
|
index.contains_expressions
|
|
and not self.connection.features.supports_expression_indexes
|
|
):
|
|
return None
|
|
# Index.create_sql returns interpolated SQL which makes params=None a
|
|
# necessity to avoid escaping attempts on execution.
|
|
self.execute(index.create_sql(model, self), params=None)
|
|
|
|
def remove_index(self, model, index):
|
|
"""Remove an index from a model."""
|
|
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):
|
|
"""Add a constraint to a model."""
|
|
sql = constraint.create_sql(model, self)
|
|
if sql:
|
|
# Constraint.create_sql returns interpolated SQL which makes
|
|
# params=None a necessity to avoid escaping attempts on execution.
|
|
self.execute(sql, params=None)
|
|
|
|
def remove_constraint(self, model, constraint):
|
|
"""Remove a constraint from a model."""
|
|
sql = constraint.remove_sql(model, self)
|
|
if sql:
|
|
self.execute(sql)
|
|
|
|
def alter_unique_together(self, model, old_unique_together, new_unique_together):
|
|
"""
|
|
Deal with a model changing its unique_together. The input
|
|
unique_togethers must be doubly-nested, not the single-nested
|
|
["foo", "bar"] format.
|
|
"""
|
|
olds = {tuple(fields) for fields in old_unique_together}
|
|
news = {tuple(fields) for fields in new_unique_together}
|
|
# Deleted uniques
|
|
for fields in olds.difference(news):
|
|
self._delete_composed_index(
|
|
model,
|
|
fields,
|
|
{"unique": True, "primary_key": False},
|
|
self.sql_delete_unique,
|
|
)
|
|
# Created uniques
|
|
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):
|
|
"""
|
|
Deal with a model changing its index_together. The input
|
|
index_togethers must be doubly-nested, not the single-nested
|
|
["foo", "bar"] format.
|
|
"""
|
|
olds = {tuple(fields) for fields in old_index_together}
|
|
news = {tuple(fields) for fields in new_index_together}
|
|
# Deleted indexes
|
|
for fields in olds.difference(news):
|
|
self._delete_composed_index(
|
|
model,
|
|
fields,
|
|
{"index": True, "unique": False},
|
|
self.sql_delete_index,
|
|
)
|
|
# Created indexes
|
|
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
|
|
):
|
|
# Constraint matching the unique_together name.
|
|
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):
|
|
"""Rename the table a model points to."""
|
|
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),
|
|
}
|
|
)
|
|
# Rename all references to the old table name.
|
|
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):
|
|
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):
|
|
"""Move a model's table between tablespaces."""
|
|
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):
|
|
"""
|
|
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
|
|
if field.many_to_many and field.remote_field.through._meta.auto_created:
|
|
return self.create_model(field.remote_field.through)
|
|
# Get the column's definition
|
|
definition, params = self.column_sql(model, field, include_default=True)
|
|
# It might not actually have a column behind it
|
|
if definition is None:
|
|
return
|
|
if col_type_suffix := field.db_type_suffix(connection=self.connection):
|
|
definition += f" {col_type_suffix}"
|
|
# Check constraints can go on the column SQL here
|
|
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"
|
|
# Add FK constraint inline, if supported.
|
|
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(),
|
|
}
|
|
# Otherwise, add FK constraints later.
|
|
else:
|
|
self.deferred_sql.append(
|
|
self._create_fk_sql(model, field, constraint_suffix)
|
|
)
|
|
# Build the SQL and run it
|
|
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)
|
|
# Drop the default if we need to
|
|
# (Django usually does not use in-database defaults)
|
|
if (
|
|
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)
|
|
# Add field comment, if required.
|
|
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
|
|
)
|
|
)
|
|
# Add an index, if required
|
|
self.deferred_sql.extend(self._field_indexes_sql(model, field))
|
|
# Reset connection if required
|
|
if self.connection.features.connection_persists_old_columns:
|
|
self.connection.close()
|
|
|
|
def remove_field(self, model, field):
|
|
"""
|
|
Remove a field from a model. Usually involves deleting a column,
|
|
but for M2Ms may involve deleting a table.
|
|
"""
|
|
# Special-case implicit M2M tables
|
|
if field.many_to_many and field.remote_field.through._meta.auto_created:
|
|
return self.delete_model(field.remote_field.through)
|
|
# It might not actually have a column behind it
|
|
if field.db_parameters(connection=self.connection)["type"] is None:
|
|
return
|
|
# Drop any FK constraints, MySQL requires explicit deletion
|
|
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))
|
|
# Delete the column
|
|
sql = self.sql_delete_column % {
|
|
"table": self.quote_name(model._meta.db_table),
|
|
"column": self.quote_name(field.column),
|
|
}
|
|
self.execute(sql)
|
|
# Reset connection if required
|
|
if self.connection.features.connection_persists_old_columns:
|
|
self.connection.close()
|
|
# Remove all deferred statements referencing the deleted column.
|
|
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):
|
|
"""
|
|
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.
|
|
"""
|
|
if not self._field_should_be_altered(old_field, new_field):
|
|
return
|
|
# Ensure this field is even column-based
|
|
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
|
|
)
|
|
):
|
|
# Both sides have through models; this is a no-op.
|
|
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)
|
|
)
|
|
|
|
self._alter_field(
|
|
model,
|
|
old_field,
|
|
new_field,
|
|
old_type,
|
|
new_type,
|
|
old_db_params,
|
|
new_db_params,
|
|
strict,
|
|
)
|
|
|
|
def _alter_field(
|
|
self,
|
|
model,
|
|
old_field,
|
|
new_field,
|
|
old_type,
|
|
new_type,
|
|
old_db_params,
|
|
new_db_params,
|
|
strict=False,
|
|
):
|
|
"""Perform a "physical" (non-ManyToMany) field update."""
|
|
# Drop any FK constraints, we'll remake them later
|
|
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))
|
|
# Has unique been removed?
|
|
if old_field.unique and (
|
|
not new_field.unique or self._field_became_primary_key(old_field, new_field)
|
|
):
|
|
# Find the unique constraint for this field
|
|
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))
|
|
# 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.
|
|
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:
|
|
# '_meta.related_field' also contains M2M reverse fields, these
|
|
# will be filtered out
|
|
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))
|
|
# 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
|
|
if (
|
|
old_field.db_index
|
|
and not old_field.unique
|
|
and (not new_field.db_index or new_field.unique)
|
|
):
|
|
# Find the index for this field
|
|
meta_index_names = {index.name for index in model._meta.indexes}
|
|
# Retrieve only BTREE indexes since this is what's created with
|
|
# db_index=True.
|
|
index_names = self._constraint_names(
|
|
model,
|
|
[old_field.column],
|
|
index=True,
|
|
type_=Index.suffix,
|
|
exclude=meta_index_names,
|
|
)
|
|
for index_name in index_names:
|
|
# The only way to check if an index was created with
|
|
# db_index=True or with Index(['field'], name='foo')
|
|
# is to look at its name (refs #28053).
|
|
self.execute(self._delete_index_sql(model, index_name))
|
|
# Change check constraints?
|
|
if old_db_params["check"] != new_db_params["check"] and old_db_params["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))
|
|
# Have they renamed the column?
|
|
if old_field.column != new_field.column:
|
|
self.execute(
|
|
self._rename_field_sql(
|
|
model._meta.db_table, old_field, new_field, new_type
|
|
)
|
|
)
|
|
# Rename all references to the renamed column.
|
|
for sql in self.deferred_sql:
|
|
if isinstance(sql, Statement):
|
|
sql.rename_column_references(
|
|
model._meta.db_table, old_field.column, new_field.column
|
|
)
|
|
# Next, start accumulating actions to do
|
|
actions = []
|
|
null_actions = []
|
|
post_actions = []
|
|
# Type suffix change? (e.g. auto increment).
|
|
old_type_suffix = old_field.db_type_suffix(connection=self.connection)
|
|
new_type_suffix = new_field.db_type_suffix(connection=self.connection)
|
|
# Type, collation, or comment change?
|
|
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)
|
|
# 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?
|
|
needs_database_default = False
|
|
if old_field.null and not new_field.null:
|
|
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)
|
|
)
|
|
# Nullability change?
|
|
if old_field.null != new_field.null:
|
|
fragment = self._alter_column_null_sql(model, old_field, new_field)
|
|
if fragment:
|
|
null_actions.append(fragment)
|
|
# Only if we have a default and there is a change from NULL to NOT NULL
|
|
four_way_default_alteration = new_field.has_default() and (
|
|
old_field.null and not new_field.null
|
|
)
|
|
if actions or null_actions:
|
|
if not four_way_default_alteration:
|
|
# If we don't have to do a 4-way default alteration we can
|
|
# directly run a (NOT) NULL alteration
|
|
actions += null_actions
|
|
# Combine actions together if we can (e.g. postgres)
|
|
if self.connection.features.supports_combined_alters and actions:
|
|
sql, params = tuple(zip(*actions))
|
|
actions = [(", ".join(sql), sum(params, []))]
|
|
# Apply those actions
|
|
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:
|
|
# Update existing rows with default value
|
|
self.execute(
|
|
self.sql_update_with_default
|
|
% {
|
|
"table": self.quote_name(model._meta.db_table),
|
|
"column": self.quote_name(new_field.column),
|
|
"default": "%s",
|
|
},
|
|
[new_default],
|
|
)
|
|
# Since we didn't run a NOT NULL change before we need to do it
|
|
# now
|
|
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 primary_key changed to False, delete the primary key constraint.
|
|
if old_field.primary_key and not new_field.primary_key:
|
|
self._delete_primary_key(model, strict)
|
|
# Added a unique?
|
|
if self._unique_should_be_added(old_field, new_field):
|
|
self.execute(self._create_unique_sql(model, [new_field]))
|
|
# 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
|
|
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]))
|
|
# Type alteration on primary key? Then we need to alter the column
|
|
# referring to us.
|
|
rels_to_update = []
|
|
if drop_foreign_keys:
|
|
rels_to_update.extend(_related_non_m2m_objects(old_field, new_field))
|
|
# Changed to become primary key?
|
|
if self._field_became_primary_key(old_field, new_field):
|
|
# Make the new one
|
|
self.execute(self._create_primary_key_sql(model, new_field))
|
|
# Update all referencing columns
|
|
rels_to_update.extend(_related_non_m2m_objects(old_field, new_field))
|
|
# Handle our type alters on the other end of rels from the PK stuff above
|
|
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)
|
|
# Does it have a foreign key?
|
|
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")
|
|
)
|
|
# Rebuild FKs that pointed to us if we previously had to drop them
|
|
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")
|
|
)
|
|
# Does it have check constraints we need to add?
|
|
if old_db_params["check"] != new_db_params["check"] and new_db_params["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"])
|
|
)
|
|
# Drop the default if we need to
|
|
# (Django usually does not use in-database defaults)
|
|
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)
|
|
# Reset connection if required
|
|
if self.connection.features.connection_persists_old_columns:
|
|
self.connection.close()
|
|
|
|
def _alter_column_null_sql(self, model, old_field, new_field):
|
|
"""
|
|
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.
|
|
"""
|
|
if (
|
|
self.connection.features.interprets_empty_strings_as_nulls
|
|
and new_field.empty_strings_allowed
|
|
):
|
|
# The field is nullable in the database anyway, leave it alone.
|
|
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):
|
|
"""
|
|
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.
|
|
"""
|
|
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:
|
|
# Some databases (Oracle) can't take defaults as a parameter
|
|
# If this is the case, the SchemaEditor for that database should
|
|
# implement prepare_default().
|
|
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_type_sql(
|
|
self, model, old_field, new_field, new_type, old_collation, new_collation
|
|
):
|
|
"""
|
|
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 two-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.
|
|
"""
|
|
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 change?
|
|
comment_sql = ""
|
|
if self.connection.features.supports_comments and not new_field.many_to_many:
|
|
if old_field.db_comment != new_field.db_comment:
|
|
# PostgreSQL and Oracle can't execute 'ALTER COLUMN ...' and
|
|
# 'COMMENT ON ...' at the same time.
|
|
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):
|
|
"""Alter M2Ms to repoint their to= endpoints."""
|
|
# Rename the through table
|
|
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,
|
|
)
|
|
# Repoint the FK to the other side
|
|
self.alter_field(
|
|
new_field.remote_field.through,
|
|
# 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).
|
|
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,
|
|
# for self-referential models we need to alter field from the other end too
|
|
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=""):
|
|
"""
|
|
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.
|
|
"""
|
|
_, 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
|
|
# If everything fits into max_length, use that name.
|
|
index_name = "%s_%s_%s" % (table_name, "_".join(column_names), hash_suffix_part)
|
|
if len(index_name) <= max_length:
|
|
return index_name
|
|
# Shorten a long suffix.
|
|
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,
|
|
)
|
|
# Prepend D if needed to prevent the name from starting with an
|
|
# underscore or a number (not permitted on Oracle).
|
|
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,
|
|
):
|
|
"""
|
|
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, ...).
|
|
"""
|
|
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):
|
|
"""
|
|
Return a list of all index SQL statements (field indexes,
|
|
index_together, Meta.indexes) for the specified model.
|
|
"""
|
|
if not model._meta.managed or model._meta.proxy or model._meta.swapped:
|
|
return []
|
|
output = []
|
|
for field in model._meta.local_fields:
|
|
output.extend(self._field_indexes_sql(model, field))
|
|
|
|
for field_names in model._meta.index_together:
|
|
fields = [model._meta.get_field(field) for field in field_names]
|
|
output.append(self._create_index_sql(model, fields=fields, suffix="_idx"))
|
|
|
|
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):
|
|
"""
|
|
Return a list of all index SQL statements for the specified 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):
|
|
ignore = ignore or set()
|
|
_, old_path, old_args, old_kwargs = old_field.deconstruct()
|
|
_, new_path, new_args, new_kwargs = new_field.deconstruct()
|
|
# 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
|
|
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)
|
|
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_sql(
|
|
self,
|
|
model,
|
|
fields,
|
|
name,
|
|
condition=None,
|
|
deferrable=None,
|
|
include=None,
|
|
opclasses=None,
|
|
expressions=None,
|
|
):
|
|
if (
|
|
deferrable
|
|
and not self.connection.features.supports_deferrable_unique_constraints
|
|
):
|
|
return None
|
|
if condition or include or opclasses or expressions:
|
|
# Databases support conditional, covering, and functional unique
|
|
# constraints via a unique index.
|
|
sql = self._create_unique_sql(
|
|
model,
|
|
fields,
|
|
name=name,
|
|
condition=condition,
|
|
include=include,
|
|
opclasses=opclasses,
|
|
expressions=expressions,
|
|
)
|
|
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,
|
|
):
|
|
if (
|
|
(
|
|
deferrable
|
|
and not self.connection.features.supports_deferrable_unique_constraints
|
|
)
|
|
or (condition and not self.connection.features.supports_partial_indexes)
|
|
or (include and not self.connection.features.supports_covering_indexes)
|
|
or (
|
|
expressions and not self.connection.features.supports_expression_indexes
|
|
)
|
|
):
|
|
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),
|
|
)
|
|
|
|
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,
|
|
):
|
|
if (
|
|
(
|
|
deferrable
|
|
and not self.connection.features.supports_deferrable_unique_constraints
|
|
)
|
|
or (condition and not self.connection.features.supports_partial_indexes)
|
|
or (include and not self.connection.features.supports_covering_indexes)
|
|
or (
|
|
expressions and not self.connection.features.supports_expression_indexes
|
|
)
|
|
):
|
|
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):
|
|
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,
|
|
):
|
|
"""Return all constraint names matching the columns and conditions."""
|
|
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)
|