python-django-evolution 0.6.7-1 / usr / share / pyshared / django_evolution / mutations.py

import copy

from django.db.models.fields import *
from django.db.models.fields.related import *
from django.db import models
from django.utils.datastructures import SortedDict
from django.utils.functional import curry

from django_evolution.signature import ATTRIBUTE_DEFAULTS
from django_evolution import CannotSimulate, SimulationFailure, EvolutionNotImplementedError, is_multi_db
from django_evolution.db import EvolutionOperationsMulti

FK_INTEGER_TYPES = [
    'AutoField', 'PositiveIntegerField', 'PositiveSmallIntegerField'
]

if is_multi_db():
    from django.db import router


def create_field(proj_sig, field_name, field_type, field_attrs, parent_model):
    """
    Create an instance of a field from a field signature. This is useful for
    accessing all the database property mechanisms built into fields.
    """
    # related_model isn't a valid field attribute, so it must be removed
    # prior to instantiating the field, but it must be restored
    # to keep the signature consistent.
    related_model = field_attrs.pop('related_model', None)

    if related_model:
        related_app_name, related_model_name = related_model.split('.')
        related_model_sig = proj_sig[related_app_name][related_model_name]
        to = MockModel(proj_sig, related_app_name, related_model_name,
                       related_model_sig, stub=True)

        field = field_type(to, name=field_name, **field_attrs)
        field_attrs['related_model'] = related_model
    else:
        field = field_type(name=field_name, **field_attrs)

    if field_type == ManyToManyField and parent_model is not None:
        # Starting in Django 1.2, a ManyToManyField must have a through
        # model defined. This will be set internally to an auto-created
        # model if one isn't specified. We have to fake that model.
        through_model = field_attrs.get('through_model', None)
        through_model_sig = None

        if through_model:
            through_app_name, through_model_name = through_model.split('.')
            through_model_sig = proj_sig[through_app_name][through_model_name]
        elif hasattr(field, '_get_m2m_attr'):
            # Django >= 1.2
            to = field.rel.to._meta.object_name.lower()

            if (field.rel.to == RECURSIVE_RELATIONSHIP_CONSTANT or
                to == parent_model._meta.object_name.lower()):
                from_ = 'from_%s' % to
                to = 'to_%s' % to
            else:
                from_ = parent_model._meta.object_name.lower()

            # This corresponds to the signature in
            # related.create_many_to_many_intermediary_model
            through_app_name = parent_model.app_name
            through_model_name = '%s_%s' % (parent_model._meta.object_name,
                                            field.name)
            through_model = '%s.%s' % (through_app_name, through_model_name)

            fields = SortedDict()
            fields['id'] = {
                'field_type': AutoField,
                'primary_key': True,
            }

            fields[from_] = {
                'field_type': ForeignKey,
                'related_model': '%s.%s' % (parent_model.app_name,
                                            parent_model._meta.object_name),
                'related_name': '%s+' % through_model_name,
            }

            fields[to] = {
                'field_type': ForeignKey,
                'related_model': related_model,
                'related_name': '%s+' % through_model_name,
            }

            through_model_sig = {
                'meta': {
                    'db_table': field._get_m2m_db_table(parent_model._meta),
                    'managed': True,
                    'auto_created': True,
                    'app_label': through_app_name,
                    'unique_together': ((from_, to),),
                    'pk_column': 'id',
                },
                'fields': fields,
            }

            field.auto_created = True

        if through_model_sig:
            through = MockModel(proj_sig, through_app_name, through_model_name,
                                through_model_sig)
            field.rel.through = through

        field.m2m_db_table = curry(field._get_m2m_db_table, parent_model._meta)
        field.set_attributes_from_rel()

    field.set_attributes_from_name(field_name)

    return field


class MockMeta(object):
    """
    A mockup of a models Options object, based on the model signature.

    The stub argument is used to circumvent recursive relationships. If
    'stub' is provided, the constructed model will only be a stub -
    it will only have a primary key field.
    """
    def __init__(self, proj_sig, app_name, model_name, model_sig):
        self.object_name = model_name
        self.app_label = app_name
        self.meta = {
            'order_with_respect_to': None,
            'has_auto_field': None,
            'db_tablespace': None,
        }
        self.meta.update(model_sig['meta'])
        self._fields = SortedDict()
        self._many_to_many = SortedDict()
        self.abstract = False
        self.managed = True
        self.proxy = False
        self._model_sig = model_sig
        self._proj_sig = proj_sig

    def setup_fields(self, model, stub=False):
        for field_name, field_sig in self._model_sig['fields'].items():
            if not stub or field_sig.get('primary_key', False):
                field_type = field_sig.pop('field_type')
                field = create_field(self._proj_sig, field_name, field_type,
                                     field_sig, model)

                if AutoField == type(field):
                    self.meta['has_auto_field'] = True
                    self.meta['auto_field'] = field

                field_sig['field_type'] = field_type

                if ManyToManyField == type(field):
                    self._many_to_many[field.name] = field
                else:
                    self._fields[field.name] = field

                field.set_attributes_from_name(field_name)
                if field_sig.get('primary_key', False):
                    self.pk = field

    def __getattr__(self, name):
        return self.meta[name]

    def get_field(self, name):
        try:
            return self._fields[name]
        except KeyError:
            try:
                return self._many_to_many[name]
            except KeyError:
                raise FieldDoesNotExist('%s has no field named %r' %
                                        (self.object_name, name))

    def get_field_by_name(self, name):
        return (self.get_field(name), None, True, None)

    def get_fields(self):
        return self._fields.values()

    def get_many_to_many_fields(self):
        return self._many_to_many.values()

    fields = property(fget=get_fields)
    local_fields = property(fget=get_fields)
    local_many_to_many = property(fget=get_many_to_many_fields)


class MockModel(object):
    """
    A mockup of a model object, providing sufficient detail
    to derive database column and table names using the standard
    Django fields.
    """
    def __init__(self, proj_sig, app_name, model_name, model_sig, stub=False):
        self.app_name = app_name
        self.model_name = model_name
        self._meta = MockMeta(proj_sig, app_name, model_name, model_sig)
        self._meta.setup_fields(self, stub)

    def __eq__(self, other):
        # For our purposes, we don't want to appear equal to "self".
        # Really, Django 1.2 should be checking if this is a string before
        # doing this comparison,
        return (isinstance(other, MockModel) and
                self.app_name == other.app_name and
                self.model_name == other.model_name)


class MockRelated(object):
    """
    A mockup of django.db.models.related.RelatedObject, providing
    sufficient detail to derive database column and table names using
    the standard Django fields.
    """
    def __init__(self, related_model, model, field):
        self.parent_model = related_model
        self.model = model
        self.opts = model._meta
        self.field = field
        self.name = '%s:%s' % (model.app_name, model.model_name)
        self.var_name = model.model_name.lower()


class BaseMutation:
    def __init__(self):
        pass

    def mutate(self, app_label, proj_sig, target_database=None):
        """
        Performs the mutation on the database. Database changes will occur
        after this function is invoked.
        """
        raise NotImplementedError()

    def simulate(self, app_label, proj_sig, target_database=None):
        """
        Performs a simulation of the mutation to be performed. The purpose of
        the simulate function is to ensure that after all mutations have occured
        the database will emerge in a state consistent with the currently loaded
        models file.
        """
        raise NotImplementedError()

    def is_mutable(self, app_label, proj_sig, database):
        """
        test if the current mutation could be applied to the given database
        """
        return False


class MonoBaseMutation(BaseMutation):
    # introducting model_name at this stage will prevent subclasses to be
    # cross databases
    def __init__(self, model_name = None):
        BaseMutation.__init__(self)
        self.model_name = model_name

    def evolver(self, model, database=None):
        if is_multi_db() and database is None:
            db_name = router.db_for_write(model)
        else:
            db_name = database or 'default'

        return EvolutionOperationsMulti(db_name).get_evolver()

    def is_mutable(self, app_label, proj_sig, database):
        if is_multi_db():
            app_sig = proj_sig[app_label]
            model_sig = app_sig[self.model_name]
            model = MockModel(proj_sig, app_label, self.model_name, model_sig)
            db_name = router.db_for_write(model)
            return db_name and db_name == database
        else:
            return True


class SQLMutation(BaseMutation):
    def __init__(self, tag, sql, update_func=None):
        self.tag = tag
        self.sql = sql
        self.update_func = update_func

    def __str__(self):
        return "SQLMutation('%s')" % self.tag

    def simulate(self, app_label, proj_sig, database=None):
        """SQL mutations cannot be simulated unless an update function is
        provided"""

        if callable(self.update_func):
            self.update_func(app_label, proj_sig)
        else:
            raise CannotSimulate('Cannot simulate SQLMutations')

    def mutate(self, app_label, proj_sig, database=None):
        "The mutation of an SQL mutation returns the raw SQL"
        return self.sql

    def is_mutable(self, app_label, proj_sig, database):
        return True


class DeleteField(MonoBaseMutation):
    def __init__(self, model_name, field_name):
        MonoBaseMutation.__init__(self, model_name)
        self.field_name = field_name

    def __str__(self):
        return "DeleteField('%s', '%s')" % (self.model_name, self.field_name)

    def simulate(self, app_label, proj_sig, database=None):
        app_sig = proj_sig[app_label]
        model_sig = app_sig[self.model_name]

        # If the field was used in the unique_together attribute, update it.
        unique_together = model_sig['meta']['unique_together']
        unique_together_list = []

        for ut_index in range(0, len(unique_together), 1):
            ut = unique_together[ut_index]
            unique_together_fields = []

            for field_name_index in range(0, len(ut), 1):
                field_name = ut[field_name_index]

                if not field_name == self.field_name:
                    unique_together_fields.append(field_name)

            unique_together_list.append(tuple(unique_together_fields))

        model_sig['meta']['unique_together'] = tuple(unique_together_list)

        if model_sig['fields'][self.field_name].get('primary_key',False):
            raise SimulationFailure('Cannot delete a primary key.')

        # Simulate the deletion of the field.
        try:
            model_sig['fields'].pop(self.field_name)
        except KeyError:
            raise SimulationFailure('Cannot find the field named "%s".'
                                    % self.field_name)

    def mutate(self, app_label, proj_sig, database=None):
        app_sig = proj_sig[app_label]
        model_sig = app_sig[self.model_name]
        field_sig = model_sig['fields'][self.field_name]

        model = MockModel(proj_sig, app_label, self.model_name, model_sig)

        # Temporarily remove field_type from the field signature
        # so that we can create a field
        field_type = field_sig.pop('field_type')
        field = create_field(proj_sig, self.field_name, field_type, field_sig,
                             model)
        field_sig['field_type'] = field_type

        evolver = self.evolver(model, database)

        if field_type == models.ManyToManyField:
            sql_statements = \
                evolver.delete_table(field._get_m2m_db_table(model._meta))
        else:
            sql_statements = evolver.delete_column(model, field)

        return sql_statements


class AddField(MonoBaseMutation):
    def __init__(self, model_name, field_name, field_type,
                 initial=None, **kwargs):
        MonoBaseMutation.__init__(self, model_name)
        self.field_name = field_name
        self.field_type = field_type
        self.field_attrs = kwargs
        self.initial = initial

    def __str__(self):
        if self.field_type.__module__.startswith('django.db.models'):
            field_prefix = 'models.'
        else:
            field_prefix = ''

        str_output = ["'%(model_name)s', '%(field_name)s', %(field_type)s" % {
            'model_name': self.model_name,
            'field_name': self.field_name,
            'field_type': field_prefix + self.field_type.__name__
        }]

        if self.initial is not None:
            str_output.append('initial=%s' % repr(self.initial))

        for key,value in self.field_attrs.items():
            str_output.append("%s=%s" % (key,repr(value)))

        return 'AddField(' + ', '.join(str_output) + ')'

    def simulate(self, app_label, proj_sig, database=None):
        app_sig = proj_sig[app_label]
        model_sig = app_sig[self.model_name]

        if self.field_name in model_sig['fields']:
            raise SimulationFailure(
                "Model '%s.%s' already has a field named '%s'"
                % (app_label, self.model_name, self.field_name))

        if (self.field_type != models.ManyToManyField and
            not self.field_attrs.get('null', ATTRIBUTE_DEFAULTS['null'])
            and self.initial is None):
            raise SimulationFailure(
                "Cannot create new column '%s' on '%s.%s' without a "
                "non-null initial value."
                % (self.field_name, app_label, self.model_name))

        model_sig['fields'][self.field_name] = {
            'field_type': self.field_type,
        }

        model_sig['fields'][self.field_name].update(self.field_attrs)

    def mutate(self, app_label, proj_sig, database=None):
        if self.field_type == models.ManyToManyField:
            return self.add_m2m_table(app_label, proj_sig, database)
        else:
            return self.add_column(app_label, proj_sig, database)

    def add_column(self, app_label, proj_sig, database):
        app_sig = proj_sig[app_label]
        model_sig = app_sig[self.model_name]

        model = MockModel(proj_sig, app_label, self.model_name, model_sig)
        field = create_field(proj_sig, self.field_name, self.field_type,
                             self.field_attrs, model)

        evolver = self.evolver(model, database)

        sql_statements = evolver.add_column(model, field, self.initial)

        # Create SQL index if necessary
        sql_statements.extend(evolver.create_index(model, field))

        return sql_statements

    def add_m2m_table(self, app_label, proj_sig, database):
        app_sig = proj_sig[app_label]
        model_sig = app_sig[self.model_name]

        model = MockModel(proj_sig, app_label, self.model_name, model_sig)

        field = create_field(proj_sig, self.field_name, self.field_type,
                             self.field_attrs, model)

        related_app_label, related_model_name = \
            self.field_attrs['related_model'].split('.')
        related_sig = proj_sig[related_app_label][related_model_name]
        related_model = MockModel(proj_sig, related_app_label,
                                  related_model_name, related_sig)
        related = MockRelated(related_model, model, field)

        if hasattr(field, '_get_m2m_column_name'):
            # Django < 1.2
            field.m2m_column_name = curry(field._get_m2m_column_name, related)
            field.m2m_reverse_name = curry(field._get_m2m_reverse_name, related)
        else:
            # Django >= 1.2
            field.m2m_column_name = curry(field._get_m2m_attr,
                                          related, 'column')
            field.m2m_reverse_name = curry(field._get_m2m_reverse_attr,
                                           related, 'column')

        evolver = self.evolver(model, database)
        sql_statements = evolver.add_m2m_table(model, field)

        return sql_statements


class RenameField(MonoBaseMutation):
    def __init__(self, model_name, old_field_name, new_field_name,
                 db_column=None, db_table=None):
        MonoBaseMutation.__init__(self, model_name)
        self.old_field_name = old_field_name
        self.new_field_name = new_field_name
        self.db_column = db_column
        self.db_table = db_table

    def __str__(self):
        params = "'%s', '%s', '%s'" % (self.model_name, self.old_field_name,
                                       self.new_field_name)

        if self.db_column:
            params = params + ", db_column='%s'" % (self.db_column)
        if self.db_table:
            params = params + ", db_table='%s'" % (self.db_table)

        return "RenameField(%s)" % params

    def simulate(self, app_label, proj_sig, database=None):
        app_sig = proj_sig[app_label]
        model_sig = app_sig[self.model_name]
        field_dict = model_sig['fields']
        field_sig = field_dict[self.old_field_name]

        if models.ManyToManyField == field_sig['field_type']:
            if self.db_table:
                field_sig['db_table'] = self.db_table
            else:
                field_sig.pop('db_table',None)
        elif self.db_column:
            field_sig['db_column'] = self.db_column
        else:
            # db_column and db_table were not specified (or not specified for
            # the appropriate field types). Clear the old value if one was set.
            # This amounts to resetting the column or table name to the Django
            # default name
            field_sig.pop('db_column', None)

        field_dict[self.new_field_name] = field_dict.pop(self.old_field_name)

    def mutate(self, app_label, proj_sig, database=None):
        app_sig = proj_sig[app_label]
        model_sig = app_sig[self.model_name]
        old_field_sig = model_sig['fields'][self.old_field_name]

        # Temporarily remove the field type so that we can create mock field
        # instances
        field_type = old_field_sig.pop('field_type')

        # Duplicate the old field sig, and apply the table/column changes
        new_field_sig = copy.copy(old_field_sig)

        if models.ManyToManyField == field_type:
            if self.db_table:
                new_field_sig['db_table'] = self.db_table
            else:
                new_field_sig.pop('db_table', None)
        elif self.db_column:
            new_field_sig['db_column'] = self.db_column
        else:
            new_field_sig.pop('db_column', None)

        # Create the mock field instances.
        old_field = create_field(proj_sig, self.old_field_name, field_type,
                                 old_field_sig, None)
        new_field = create_field(proj_sig, self.new_field_name, field_type,
                                 new_field_sig, None)

        # Restore the field type to the signature
        old_field_sig['field_type'] = field_type

        model = MockModel(proj_sig, app_label, self.model_name, model_sig)

        evolver = self.evolver(model, database)

        if models.ManyToManyField == field_type:
            old_m2m_table = old_field._get_m2m_db_table(model._meta)
            new_m2m_table = new_field._get_m2m_db_table(model._meta)

            return evolver.rename_table(model, old_m2m_table, new_m2m_table)
        else:
            return evolver.rename_column(model._meta, old_field, new_field)


class ChangeField(MonoBaseMutation):
    def __init__(self, model_name, field_name, initial=None, **kwargs):
        MonoBaseMutation.__init__(self, model_name)
        self.field_name = field_name
        self.field_attrs = kwargs
        self.initial = initial

    def __str__(self):
        params = (self.model_name, self.field_name)
        str_output = ["'%s', '%s'" % params]

        str_output.append('initial=%s' % repr(self.initial))

        for attr_name, attr_value in self.field_attrs.items():
            if str == type(attr_value):
                str_attr_value = "'%s'" % attr_value
            else:
                str_attr_value = str(attr_value)

            str_output.append('%s=%s' % (attr_name, str_attr_value,))

        return 'ChangeField(' + ', '.join(str_output) + ')'

    def simulate(self, app_label, proj_sig, database=None):
        app_sig = proj_sig[app_label]
        model_sig = app_sig[self.model_name]
        field_sig = model_sig['fields'][self.field_name]

        # Catch for no-op changes.
        for field_attr, attr_value in self.field_attrs.items():
            field_sig[field_attr] = attr_value

        if ('null' in self.field_attrs and
            field_sig['field_type'] != models.ManyToManyField and
            not self.field_attrs['null'] and
            self.initial is None):
            raise SimulationFailure(
                "Cannot change column '%s' on '%s.%s' without a "
                "non-null initial value."
                % (self.field_name, app_label, self.model_name))

    def mutate(self, app_label, proj_sig, database=None):
        app_sig = proj_sig[app_label]
        model_sig = app_sig[self.model_name]
        old_field_sig = model_sig['fields'][self.field_name]
        model = MockModel(proj_sig, app_label, self.model_name, model_sig)

        sql_statements = []

        for field_attr, attr_value in self.field_attrs.items():
            old_field_attr = old_field_sig.get(field_attr,
                                               ATTRIBUTE_DEFAULTS[field_attr])

            # Avoid useless SQL commands if nothing has changed.
            if not old_field_attr == attr_value:
                try:
                    evolver_func = getattr(self.evolver(model, database),
                                           'change_%s' % field_attr)
                    if field_attr == 'null':
                        sql_statements.extend(
                            evolver_func(model, self.field_name, attr_value,
                            self.initial))
                    elif field_attr == 'db_table':
                        sql_statements.extend(
                            evolver_func(model, old_field_attr, attr_value))
                    else:
                        sql_statements.extend(
                            evolver_func(model, self.field_name, attr_value))
                except AttributeError:
                    raise EvolutionNotImplementedError(
                        "ChangeField does not support modifying the '%s' "
                        "attribute on '%s.%s'."
                        % (field_attr, self.model_name, self.field_name))

        return sql_statements


class DeleteModel(MonoBaseMutation):
    def __init__(self, model_name):
        MonoBaseMutation.__init__(self, model_name)

    def __str__(self):
        return "DeleteModel(%r)" % self.model_name

    def simulate(self, app_label, proj_sig, database=None):
        app_sig = proj_sig[app_label]

        # Simulate the deletion of the model.
        del app_sig[self.model_name]

    def mutate(self, app_label, proj_sig, database=None):
        app_sig = proj_sig[app_label]
        model_sig = app_sig[self.model_name]

        sql_statements = []
        model = MockModel(proj_sig, app_label, self.model_name, model_sig)
        evolver = self.evolver(model, database)

        # Remove any many to many tables.
        for field_name, field_sig in model_sig['fields'].items():
            if field_sig['field_type'] == models.ManyToManyField:
                field = model._meta.get_field(field_name)
                m2m_table = field._get_m2m_db_table(model._meta)
                sql_statements += evolver.delete_table(m2m_table)

        # Remove the table itself.
        sql_statements += evolver.delete_table(model._meta.db_table)

        return sql_statements


class DeleteApplication(BaseMutation):
    def __str__(self):
        return 'DeleteApplication()'

    def simulate(self, app_label, proj_sig, database=None):
        if database:
            app_sig = proj_sig[app_label]

            # Simulate the deletion of the models.
            for model_name in app_sig.keys():
                mutation = DeleteModel(model_name)

                if mutation.is_mutable(app_label, proj_sig, database):
                    del app_sig[self.model_name]

    def mutate(self, app_label, proj_sig, database=None):
        sql_statements = []

        # This test will introduce a regression, but we can't afford to remove
        # all models at a same time if they aren't owned by the same database
        if database:
            app_sig = proj_sig[app_label]

            for model_name in app_sig.keys():
                mutation = DeleteModel(model_name)

                if mutation.is_mutable(app_label, proj_sig, database):
                    sql_statements.extend(mutation.mutate(app_label, proj_sig))

        return sql_statements

    def is_mutable(self, app_label, proj_sig, database):
        # the test is done in the mutate method above. We can return True
        return True