Python django.db.migrations 模块，AddField() 实例源码

def generate_operations():

    operations = []

    for code, name in settings.LANGUAGES:

        operations += [

            migrations.AddField(
                model_name='warehouse',
                name='address_%s' % code,
                field=models.CharField(db_index=True, max_length=255, null=True, verbose_name='Address'),
            ),
            migrations.AddField(
                model_name='warehouse',
                name='title_%s' % code,
                field=models.CharField(db_index=True, max_length=255, null=True, verbose_name='Title'),
            ),

        ]

        return operations

def test_add_field(self):
        project_state = self.set_up_test_model()
        operation = migrations.AddField(
            'Pony',
            'height',
            models.FloatField(null=True, default=5)
        )
        new_state = project_state.clone()
        operation.state_forwards('tests', new_state)
        self.assertColumnNotExists('tests_pony', 'height')
        with connection.schema_editor() as editor:
            operation.database_forwards('tests', editor, project_state, new_state)
        self.assertColumnExists('tests_pony', 'height')
        with connection.schema_editor() as editor:
            operation.database_backwards('tests', editor, new_state, project_state)
        self.assertColumnNotExists('tests_pony', 'height')

def test_custom_migration_operation(self):
        project_state = self.set_up_test_model()
        operation = AddField(
            app_label='tests',
            model_name='pony',
            name='yellow',
            field=models.BooleanField(default=True)
        )
        new_state = project_state.clone()
        operation.state_forwards('tests', new_state)

        self.assertColumnNotExists('tests_pony', 'yellow')
        with connection.schema_editor() as editor:
            operation.database_forwards('tests', editor, project_state, new_state)

        self.assertColumnExists('tests_pony', 'yellow')
        with connection.schema_editor() as editor:
            operation.database_backwards('tests', editor, new_state, project_state)

        self.assertColumnNotExists('tests_pony', 'yellow')

def get_add_field_for_langs(name, **kwargs):
    ret = []

    for lang_code in mt_settings.AVAILABLE_LANGUAGES:
        trans_field = build_localized_fieldname(name, lang_code)
        ret.append(migrations.AddField(name=trans_field, **kwargs))

    return ret

def __init__(self, *args, **kwargs):
        self.app_label = kwargs.pop('app_label')
        super(AddField, self).__init__(*args, **kwargs)

def state_forwards(self, app_label, state):
        return super(AddField, self).state_forwards(self.app_label, state)

def database_forwards(self, app_label, *args):
        return super(AddField, self).database_forwards(self.app_label, *args)

def add_field(field, filters: List[str]):
    """Adds the specified field to a model.

    Arguments:
        field:
            The field to add to a model.

        filters:
            List of strings to filter
            SQL statements on.
    """

    model = define_fake_model()
    project = migrations.state.ProjectState.from_apps(apps)

    with connection.schema_editor() as schema_editor:
        execute_migration(schema_editor, [
            migrations.CreateModel(
                model.__name__,
                fields=[]
            )
        ], project)

    with filtered_schema_editor(*filters) as (schema_editor, calls):
        execute_migration(schema_editor, [
            migrations.AddField(
                model.__name__,
                'title',
                field
            )
        ], project)

    yield calls

def reduce_add_field_alter_field(self, operation, other, in_between):
        if (operation.model_name_lower == other.model_name_lower and
                operation.name_lower == other.name_lower):
            return [
                migrations.AddField(
                    model_name=operation.model_name,
                    name=operation.name,
                    field=other.field,
                )
            ]

def reduce_add_field_rename_field(self, operation, other, in_between):
        if (operation.model_name_lower == other.model_name_lower and
                operation.name_lower == other.old_name_lower):
            return [
                migrations.AddField(
                    model_name=operation.model_name,
                    name=other.new_name,
                    field=operation.field,
                )
            ]

def __init__(self, name, app_label):
        # by changing app_label here to 'wagtailcore' we trick Django migrations system
        # to think that this migration belongs to wagtailcore app
        # this is necessary to make model name resolution work
        app_label = 'wagtailcore'
        super(Migration, self).__init__(name, app_label)

        # find last wagtailcore migration
        mod_name  = MigrationLoader.migrations_module(app_label)
        if DJANGO_VERSION >= (1, 11):
            # Django 1.11 returns tuple(str, bool) while older versions return str
            mod_name = mod_name[0]
        mod = import_module(mod_name)
        migrations = []
        # this loop acts the same way as MigrationLoader.
        for name in os.listdir(os.path.dirname(mod.__file__)):
            if not name.endswith('.py'):
                continue
            import_name = name.rsplit('.', 1)[0]
            if import_name[0] in '_.~':
                continue
            migrations.append(import_name)
        last_migration = sorted(migrations, reverse=True)[0]
        # By using `replaces` we make sure that this migration doesn't have ambiguous `app_label`.
        # When this migration is applied Django writes only replaced migration
        # to django_migrations table in DB. Otherwise migration would have
        # 'wagtailtranslation' as app_label in django_migrations table and
        # `migrate` command would consider this migration as unapplied due
        # to app_label mismatch.
        self.replaces = [
            (app_label, last_migration),
        ]

        # import operations from wagtail migration we are replacing
        # and prepend them to operations of this migration
        mod_path = '{}.{}'.format(mod_name, last_migration)
        orig_migration = import_module(mod_path).Migration
        self.operations[:0] = orig_migration.operations
        self.dependencies = orig_migration.dependencies

    # Dynamically define AddField operations for all Page field translations.
    # This always uses current AVAILABLE_LANGUAGES setting.
    # In case languages are changed after running this migration, `makemigrations`
    # command would do nothing for Page model. One would have to run `sync_translation_fields`
    # command from modeltranslation to get DB schema in sync.