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

def migrate(self, ops, state=None):
        class Migration(migrations.Migration):
            operations = ops
        migration = Migration('name', 'tests')
        inject_trigger_operations([(migration, False)])
        with connection.schema_editor() as schema_editor:
            return migration.apply(state or ProjectState.from_apps(self.apps), schema_editor)

def insert_notifications(apps, schema_editor):
    import dsmr_frontend.services
    Notification = apps.get_model('dsmr_frontend', 'Notification')

    # Search for any applied migrations in the past. This should indicate a long(er) living instance of the project.
    existing_project = MigrationRecorder.Migration.objects.filter(
        applied__lt=timezone.now() - timezone.timedelta(hours=24)
    ).exists()

    if existing_project:
        return

    Notification.objects.create(
        message=dsmr_frontend.services.get_translated_string(
            text=_('Welcome to DSMR-reader! Please make sure to check your settings in the Configuration page!')
        ),
        redirect_to='admin:index'
    )

    Notification.objects.create(
        message=dsmr_frontend.services.get_translated_string(
            text=_('You may check the status of your readings and data in the Status page.')
        ),
        redirect_to='frontend:status'
    )

def get_fake_model(fields=None, model_base=PostgresModel, meta_options={}):
    """Creates a fake model to use during unit tests."""

    model = define_fake_model(fields, model_base, meta_options)

    class TestProject:

        def clone(self, *_args, **_kwargs):
            return self

        @property
        def apps(self):
            return self

    class TestMigration(migrations.Migration):
        operations = [HStoreExtension()]

    with connection.schema_editor() as schema_editor:
        migration_executor = MigrationExecutor(schema_editor.connection)
        migration_executor.apply_migration(
            TestProject(), TestMigration('eh', 'postgres_extra'))

        schema_editor.create_model(model)

    return model

def get_fake_model(fields=None, model_base=PostgresModel, meta_options={}):
    """Creates a fake model to use during unit tests."""

    model = define_fake_model(fields, model_base, meta_options)

    class TestProject:

        def clone(self, *_args, **_kwargs):
            return self

        @property
        def apps(self):
            return self

    class TestMigration(migrations.Migration):
        operations = [HStoreExtension()]

    with connection.schema_editor() as schema_editor:
        migration_executor = MigrationExecutor(schema_editor.connection)
        migration_executor.apply_migration(
            TestProject(), TestMigration('eh', 'postgres_extra'))

        schema_editor.create_model(model)

    return model

def get_fake_model(fields=None, model_base=LocalizedModel, meta_options={}):
    """Creates a fake model to use during unit tests."""

    model = define_fake_model(fields, model_base, meta_options)

    class TestProject:

        def clone(self, *_args, **_kwargs):
            return self

        @property
        def apps(self):
            return self

    class TestMigration(migrations.Migration):
        operations = [HStoreExtension()]

    with connection.schema_editor() as schema_editor:
        migration_executor = MigrationExecutor(schema_editor.connection)
        migration_executor.apply_migration(
            TestProject(), TestMigration('eh', 'postgres_extra'))

        schema_editor.create_model(model)

    return model

def copy_prices(apps, schema_editor):
    """
    Validate CoursePrice then iterate over CoursePrice and populate Program.price
    """
    Program = apps.get_model('courses', 'Program')
    CourseRun = apps.get_model('courses', 'CourseRun')

    program_prices = {}
    # Note: we are not filtering on live here because all Programs will require prices and liveness won't matter

    # Validate that there is only one price per program and each run had a price. Else, fail the migration
    for run in CourseRun.objects.all():
        program = run.course.program
        # There must be one and only one price per course run
        try:
            price = run.courseprice_set.get(is_valid=True).price
        except ObjectDoesNotExist as ex:
            raise Exception("Migration failed due to a price missing for run {}".format(
                run.edx_course_key
            )) from ex
        if program.id not in program_prices:
            program_prices[program.id] = price
        elif program_prices[program.id] != price:
            raise Exception("One run in program {program} had price {price1} but another had price {price2}".format(
                program=program,
                price1=program_prices[program.id],
                price2=price,
            ))

    # Verify that all programs have prices at this point (might be false if a Program doesn't have any runs)
    for program in Program.objects.all():
        if program.id not in program_prices:
            raise Exception("Program {} does not have a price (probably due to not having any runs)".format(program))

    # Now, copy the prices
    for program_id, price in program_prices.items():
        program = Program.objects.get(id=program_id)
        program.price = price
        program.save()

def apply(self, *args, **kwargs):
        return_value = super(Migration, self).apply(*args, **kwargs)
        for user in User.objects.all():
            user.touchscreen = bool(user.skype_id)
            user.save()
        return return_value

def execute_migration(schema_editor, operations, project=None):
    """Executes the specified migration operations
    using the specified schema editor.

    Arguments:
        schema_editor:
            The schema editor to use to
            execute the migrations.

        operations:
            The migration operations to execute.

        project:
            The project state to use during the
            migrations.
    """

    project = project or migrations.state.ProjectState.from_apps(apps)

    class Migration(migrations.Migration):
        pass

    Migration.operations = operations

    executor = MigrationExecutor(schema_editor.connection)
    executor.apply_migration(
        project, Migration('eh', 'postgres_extra'))

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.