python类RunPython()的实例源码

def RenameAppInMigrationsTable(old_app_name, new_app_name):
    """
    Check whether an obsolete application name `old_app_name` is present in
    Django's `django_migrations` DB table and handle the situation as cleanly
    as possible.

    If there are migrations for the old app name, perform an UPDATE command to
    rename the app in this table so future migration runs will succeed, then
    exit with a `ObsoleteAppNameInMigrationsTableException` to indicate that
    migrations need to be re-run.

    If there are no migrations for the old app name -- e.g. the app has already
    been renamed in the table, or the old pre-rename migrations were never run
    on the DB -- then no action is performed.
    """
    return migrations.RunPython(
        _assert_and_rename_app_in_migrations(old_app_name, new_app_name)
    )

def get_operations(self):
        return '''
        # The copying of values is (sort of) reversable by a no-op:
        #  - values are copied into i18n (which is not used by anything but django-modeltrans)
        #  - the default language is copied to the orignal field, which was not used
        #    with django-modeltrans.
        migrations.RunPython(forwards, migrations.RunPython.noop),
'''

def test_run_python(self):
        """
        Because this can run arbitrary python code, we can't know
        which parts of it need to run against each schema, and which
        parts run against the public schema.

        We could hack into any generated SQL, and inspect it, looking
        for table names, attempting to push data to the correct
        schemata (including executing the SQL multiple times if
        necessary).

        Maybe we could fuck with the models generated by project_state.render(),
        and make their generated SQL do what we need it to do. Although, it looks
        like Pony.objects is a normal models.Manager class.
        """
        project_state = self.set_up_test_model()

        def forwards(models, schema_editor):
            Pony = models.get_model('tests', 'Pony')
            Pony.objects.create(pink=1, weight=3.55)
            Pony.objects.create(weight=5)

        def backwards(models, schema_editor):
            Pony = models.get_model('tests', 'Pony')
            Pony.objects.filter(pink=1, weight=3.55).delete()
            Pony.objects.filter(weight=5).delete()

        operation = migrations.RunPython(forwards, reverse_code=backwards)
        new_state = project_state.clone()
        operation.state_forwards('tests', new_state)

        Pony = project_state.apps.get_model('tests', 'Pony')

        @all_schemata
        def pony_count(count, **kwargs):
            found = Pony.objects.count()
            self.assertEqual(
                count,
                found,
                'Incorrect number of Ponies found in schema '
                '{schema}: expected {0}, found {1}'.format(count, found, **kwargs)
            )

        pony_count(0)

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

        pony_count(2)

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

        pony_count(0)