123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402 |
- from django.core.exceptions import FieldDoesNotExist
- from django.db.models.fields import NOT_PROVIDED
- from django.utils.functional import cached_property
-
- from .base import Operation
- from .utils import (
- ModelTuple, field_references_model, is_referenced_by_foreign_key,
- )
-
-
- class FieldOperation(Operation):
- def __init__(self, model_name, name, field=None):
- self.model_name = model_name
- self.name = name
- self.field = field
-
- @cached_property
- def model_name_lower(self):
- return self.model_name.lower()
-
- @cached_property
- def name_lower(self):
- return self.name.lower()
-
- def is_same_model_operation(self, operation):
- return self.model_name_lower == operation.model_name_lower
-
- def is_same_field_operation(self, operation):
- return self.is_same_model_operation(operation) and self.name_lower == operation.name_lower
-
- def references_model(self, name, app_label=None):
- name_lower = name.lower()
- if name_lower == self.model_name_lower:
- return True
- if self.field:
- return field_references_model(self.field, ModelTuple(app_label, name_lower))
- return False
-
- def references_field(self, model_name, name, app_label=None):
- model_name_lower = model_name.lower()
- # Check if this operation locally references the field.
- if model_name_lower == self.model_name_lower:
- if name == self.name:
- return True
- elif self.field and hasattr(self.field, 'from_fields') and name in self.field.from_fields:
- return True
- # Check if this operation remotely references the field.
- if self.field:
- model_tuple = ModelTuple(app_label, model_name_lower)
- remote_field = self.field.remote_field
- if remote_field:
- if (ModelTuple.from_model(remote_field.model) == model_tuple and
- (not hasattr(self.field, 'to_fields') or
- name in self.field.to_fields or None in self.field.to_fields)):
- return True
- through = getattr(remote_field, 'through', None)
- if (through and ModelTuple.from_model(through) == model_tuple and
- (getattr(remote_field, 'through_fields', None) is None or
- name in remote_field.through_fields)):
- return True
- return False
-
- def reduce(self, operation, app_label=None):
- return (
- super().reduce(operation, app_label=app_label) or
- not operation.references_field(self.model_name, self.name, app_label)
- )
-
-
- class AddField(FieldOperation):
- """Add a field to a model."""
-
- def __init__(self, model_name, name, field, preserve_default=True):
- self.preserve_default = preserve_default
- super().__init__(model_name, name, field)
-
- def deconstruct(self):
- kwargs = {
- 'model_name': self.model_name,
- 'name': self.name,
- 'field': self.field,
- }
- if self.preserve_default is not True:
- kwargs['preserve_default'] = self.preserve_default
- return (
- self.__class__.__name__,
- [],
- kwargs
- )
-
- def state_forwards(self, app_label, state):
- # If preserve default is off, don't use the default for future state
- if not self.preserve_default:
- field = self.field.clone()
- field.default = NOT_PROVIDED
- else:
- field = self.field
- state.models[app_label, self.model_name_lower].fields.append((self.name, field))
- # Delay rendering of relationships if it's not a relational field
- delay = not field.is_relation
- state.reload_model(app_label, self.model_name_lower, delay=delay)
-
- def database_forwards(self, app_label, schema_editor, from_state, to_state):
- to_model = to_state.apps.get_model(app_label, self.model_name)
- if self.allow_migrate_model(schema_editor.connection.alias, to_model):
- from_model = from_state.apps.get_model(app_label, self.model_name)
- field = to_model._meta.get_field(self.name)
- if not self.preserve_default:
- field.default = self.field.default
- schema_editor.add_field(
- from_model,
- field,
- )
- if not self.preserve_default:
- field.default = NOT_PROVIDED
-
- def database_backwards(self, app_label, schema_editor, from_state, to_state):
- from_model = from_state.apps.get_model(app_label, self.model_name)
- if self.allow_migrate_model(schema_editor.connection.alias, from_model):
- schema_editor.remove_field(from_model, from_model._meta.get_field(self.name))
-
- def describe(self):
- return "Add field %s to %s" % (self.name, self.model_name)
-
- def reduce(self, operation, app_label=None):
- if isinstance(operation, FieldOperation) and self.is_same_field_operation(operation):
- if isinstance(operation, AlterField):
- return [
- AddField(
- model_name=self.model_name,
- name=operation.name,
- field=operation.field,
- ),
- ]
- elif isinstance(operation, RemoveField):
- return []
- elif isinstance(operation, RenameField):
- return [
- AddField(
- model_name=self.model_name,
- name=operation.new_name,
- field=self.field,
- ),
- ]
- return super().reduce(operation, app_label=app_label)
-
-
- class RemoveField(FieldOperation):
- """Remove a field from a model."""
-
- def deconstruct(self):
- kwargs = {
- 'model_name': self.model_name,
- 'name': self.name,
- }
- return (
- self.__class__.__name__,
- [],
- kwargs
- )
-
- def state_forwards(self, app_label, state):
- new_fields = []
- old_field = None
- for name, instance in state.models[app_label, self.model_name_lower].fields:
- if name != self.name:
- new_fields.append((name, instance))
- else:
- old_field = instance
- state.models[app_label, self.model_name_lower].fields = new_fields
- # Delay rendering of relationships if it's not a relational field
- delay = not old_field.is_relation
- state.reload_model(app_label, self.model_name_lower, delay=delay)
-
- def database_forwards(self, app_label, schema_editor, from_state, to_state):
- from_model = from_state.apps.get_model(app_label, self.model_name)
- if self.allow_migrate_model(schema_editor.connection.alias, from_model):
- schema_editor.remove_field(from_model, from_model._meta.get_field(self.name))
-
- def database_backwards(self, app_label, schema_editor, from_state, to_state):
- to_model = to_state.apps.get_model(app_label, self.model_name)
- if self.allow_migrate_model(schema_editor.connection.alias, to_model):
- from_model = from_state.apps.get_model(app_label, self.model_name)
- schema_editor.add_field(from_model, to_model._meta.get_field(self.name))
-
- def describe(self):
- return "Remove field %s from %s" % (self.name, self.model_name)
-
- def reduce(self, operation, app_label=None):
- from .models import DeleteModel
- if isinstance(operation, DeleteModel) and operation.name_lower == self.model_name_lower:
- return [operation]
- return super().reduce(operation, app_label=app_label)
-
-
- class AlterField(FieldOperation):
- """
- Alter a field's database column (e.g. null, max_length) to the provided
- new field.
- """
-
- def __init__(self, model_name, name, field, preserve_default=True):
- self.preserve_default = preserve_default
- super().__init__(model_name, name, field)
-
- def deconstruct(self):
- kwargs = {
- 'model_name': self.model_name,
- 'name': self.name,
- 'field': self.field,
- }
- if self.preserve_default is not True:
- kwargs['preserve_default'] = self.preserve_default
- return (
- self.__class__.__name__,
- [],
- kwargs
- )
-
- def state_forwards(self, app_label, state):
- if not self.preserve_default:
- field = self.field.clone()
- field.default = NOT_PROVIDED
- else:
- field = self.field
- state.models[app_label, self.model_name_lower].fields = [
- (n, field if n == self.name else f)
- for n, f in
- state.models[app_label, self.model_name_lower].fields
- ]
- # TODO: investigate if old relational fields must be reloaded or if it's
- # sufficient if the new field is (#27737).
- # Delay rendering of relationships if it's not a relational field and
- # not referenced by a foreign key.
- delay = (
- not field.is_relation and
- not is_referenced_by_foreign_key(state, self.model_name_lower, self.field, self.name)
- )
- state.reload_model(app_label, self.model_name_lower, delay=delay)
-
- def database_forwards(self, app_label, schema_editor, from_state, to_state):
- to_model = to_state.apps.get_model(app_label, self.model_name)
- if self.allow_migrate_model(schema_editor.connection.alias, to_model):
- from_model = from_state.apps.get_model(app_label, self.model_name)
- from_field = from_model._meta.get_field(self.name)
- to_field = to_model._meta.get_field(self.name)
- if not self.preserve_default:
- to_field.default = self.field.default
- schema_editor.alter_field(from_model, from_field, to_field)
- if not self.preserve_default:
- to_field.default = NOT_PROVIDED
-
- def database_backwards(self, app_label, schema_editor, from_state, to_state):
- self.database_forwards(app_label, schema_editor, from_state, to_state)
-
- def describe(self):
- return "Alter field %s on %s" % (self.name, self.model_name)
-
- def reduce(self, operation, app_label=None):
- if isinstance(operation, RemoveField) and self.is_same_field_operation(operation):
- return [operation]
- elif isinstance(operation, RenameField) and self.is_same_field_operation(operation):
- return [
- operation,
- AlterField(
- model_name=self.model_name,
- name=operation.new_name,
- field=self.field,
- ),
- ]
- return super().reduce(operation, app_label=app_label)
-
-
- class RenameField(FieldOperation):
- """Rename a field on the model. Might affect db_column too."""
-
- def __init__(self, model_name, old_name, new_name):
- self.old_name = old_name
- self.new_name = new_name
- super().__init__(model_name, old_name)
-
- @cached_property
- def old_name_lower(self):
- return self.old_name.lower()
-
- @cached_property
- def new_name_lower(self):
- return self.new_name.lower()
-
- def deconstruct(self):
- kwargs = {
- 'model_name': self.model_name,
- 'old_name': self.old_name,
- 'new_name': self.new_name,
- }
- return (
- self.__class__.__name__,
- [],
- kwargs
- )
-
- def state_forwards(self, app_label, state):
- model_state = state.models[app_label, self.model_name_lower]
- # Rename the field
- fields = model_state.fields
- found = False
- delay = True
- for index, (name, field) in enumerate(fields):
- if not found and name == self.old_name:
- fields[index] = (self.new_name, field)
- found = True
- # Fix from_fields to refer to the new field.
- from_fields = getattr(field, 'from_fields', None)
- if from_fields:
- field.from_fields = tuple([
- self.new_name if from_field_name == self.old_name else from_field_name
- for from_field_name in from_fields
- ])
- # Delay rendering of relationships if it's not a relational
- # field and not referenced by a foreign key.
- delay = delay and (
- not field.is_relation and
- not is_referenced_by_foreign_key(state, self.model_name_lower, field, self.name)
- )
- if not found:
- raise FieldDoesNotExist(
- "%s.%s has no field named '%s'" % (app_label, self.model_name, self.old_name)
- )
- # Fix index/unique_together to refer to the new field
- options = model_state.options
- for option in ('index_together', 'unique_together'):
- if option in options:
- options[option] = [
- [self.new_name if n == self.old_name else n for n in together]
- for together in options[option]
- ]
- # Fix to_fields to refer to the new field.
- model_tuple = app_label, self.model_name_lower
- for (model_app_label, model_name), model_state in state.models.items():
- for index, (name, field) in enumerate(model_state.fields):
- remote_field = field.remote_field
- if remote_field:
- remote_model_tuple = self._get_model_tuple(
- remote_field.model, model_app_label, model_name
- )
- if remote_model_tuple == model_tuple:
- if getattr(remote_field, 'field_name', None) == self.old_name:
- remote_field.field_name = self.new_name
- to_fields = getattr(field, 'to_fields', None)
- if to_fields:
- field.to_fields = tuple([
- self.new_name if to_field_name == self.old_name else to_field_name
- for to_field_name in to_fields
- ])
- state.reload_model(app_label, self.model_name_lower, delay=delay)
-
- def database_forwards(self, app_label, schema_editor, from_state, to_state):
- to_model = to_state.apps.get_model(app_label, self.model_name)
- if self.allow_migrate_model(schema_editor.connection.alias, to_model):
- from_model = from_state.apps.get_model(app_label, self.model_name)
- schema_editor.alter_field(
- from_model,
- from_model._meta.get_field(self.old_name),
- to_model._meta.get_field(self.new_name),
- )
-
- def database_backwards(self, app_label, schema_editor, from_state, to_state):
- to_model = to_state.apps.get_model(app_label, self.model_name)
- if self.allow_migrate_model(schema_editor.connection.alias, to_model):
- from_model = from_state.apps.get_model(app_label, self.model_name)
- schema_editor.alter_field(
- from_model,
- from_model._meta.get_field(self.new_name),
- to_model._meta.get_field(self.old_name),
- )
-
- def describe(self):
- return "Rename field %s on %s to %s" % (self.old_name, self.model_name, self.new_name)
-
- def references_field(self, model_name, name, app_label=None):
- return self.references_model(model_name) and (
- name.lower() == self.old_name_lower or
- name.lower() == self.new_name_lower
- )
-
- def reduce(self, operation, app_label=None):
- if (isinstance(operation, RenameField) and
- self.is_same_model_operation(operation) and
- self.new_name_lower == operation.old_name_lower):
- return [
- RenameField(
- self.model_name,
- self.old_name,
- operation.new_name,
- ),
- ]
- # Skip `FieldOperation.reduce` as we want to run `references_field`
- # against self.new_name.
- return (
- super(FieldOperation, self).reduce(operation, app_label=app_label) or
- not operation.references_field(self.model_name, self.new_name, app_label)
- )
|