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

def set_up_test_model(self):
        operations = [
            migrations.CreateModel(
                "Pony",
                [
                    ('pony_id', models.AutoField(primary_key=True)),
                    ('pink', models.IntegerField(default=3)),
                    ('weight', models.FloatField())
                ],
            ),
            migrations.CreateModel(
                'Rider',
                [
                    ('rider_id', models.AutoField(primary_key=True)),
                    ('pony', models.ForeignKey('Pony', on_delete=models.CASCADE))
                ],
            ),
        ]
        return self.apply_operations('tests', ProjectState(), operations)

def test_demo_schemata_get_migrated(self):
        user = User.objects.create_user(**CREDENTIALS)
        schema = DemoSchema.objects.create(user=user, from_template=self.template)

        operation = migrations.CreateModel("Pony", [
            ('pony_id', models.AutoField(primary_key=True)),
            ('pink', models.IntegerField(default=1)),
        ])
        project_state = ProjectState()
        new_state = project_state.clone()
        operation.state_forwards('tests', new_state)

        schema.activate()
        self.assertFalse('tests_pony' in get_table_list())

        with connection.schema_editor() as editor:
            operation.database_forwards('tests', editor, project_state, new_state)
        schema.activate()
        self.assertTrue('tests_pony' in get_table_list())

        with connection.schema_editor() as editor:
            operation.database_backwards('tests', editor, new_state, project_state)
        schema.activate()
        self.assertFalse('tests_pony' in get_table_list())

def reduce_create_model_add_field(self, operation, other, in_between):
        if operation.name_lower == other.model_name_lower:
            # Don't allow optimizations of FKs through models they reference
            if hasattr(other.field, "rel") and other.field.rel:
                for between in in_between:
                    # Check that it doesn't point to the model
                    app_label, object_name = self.model_to_key(other.field.rel.to)
                    if between.references_model(object_name, app_label):
                        return None
                    # Check that it's not through the model
                    if getattr(other.field.rel, "through", None):
                        app_label, object_name = self.model_to_key(other.field.rel.through)
                        if between.references_model(object_name, app_label):
                            return None
            # OK, that's fine
            return [
                migrations.CreateModel(
                    operation.name,
                    fields=operation.fields + [(other.name, other.field)],
                    options=operation.options,
                    bases=operation.bases,
                    managers=operation.managers,
                )
            ]

def inject_pre_migration_operations(plan=None, apps=global_apps, using=DEFAULT_DB_ALIAS, **kwargs):
    """
    Insert a `LTreeExtension` operation before every planned `CreateModel` operation.
    """
    if plan is None:
        return

    for migration, backward in plan:
        for index, operation in enumerate(migration.operations):
            if isinstance(operation, migrations.CreateModel):
                for name, field in operation.fields:
                    if isinstance(field, LTreeField):
                        migration.operations.insert(index, LTreeExtension())
                        return

def inject_post_migration_operations(plan=None, apps=global_apps, using=DEFAULT_DB_ALIAS, **kwargs):
    if plan is None:
        return

    for migration, backward in plan:
        for index, operation in reversed(list(enumerate(migration.operations))):
            if isinstance(operation, migrations.CreateModel):
                model = apps.get_model(migration.app_label, operation.name)
                post_migrate_mpathnode(model)

def test_migrations_apply_to_templates(self):
        template = SchemaTemplate.objects.create(name='a')
        operation = migrations.CreateModel("Pony", [
            ('pony_id', models.AutoField(primary_key=True)),
            ('pink', models.IntegerField(default=1)),
        ])
        project_state = ProjectState()
        new_state = project_state.clone()
        operation.state_forwards('tests', new_state)

        template.activate()
        self.assertFalse('tests_pony' in get_table_list())

        with connection.schema_editor() as editor:
            operation.database_forwards('tests', editor, project_state, new_state)
        template.activate()
        self.assertTrue('tests_pony' in get_table_list())

        with connection.schema_editor() as editor:
            operation.database_backwards('tests', editor, new_state, project_state)
        template.activate()
        self.assertFalse('tests_pony' in get_table_list())

def create_drop_model(field, filters: List[str]):
    """Creates and drops a model with the specified field.

    Arguments:
        field:
            The field to include on the
            model to create and drop.

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

    model = define_fake_model({'title': field})

    with filtered_schema_editor(*filters) as (schema_editor, calls):
        execute_migration(schema_editor, [
            migrations.CreateModel(
                model.__name__,
                fields=[
                    ('title', field.clone())
                ]
            ),
            migrations.DeleteModel(
                model.__name__,
            )
        ])

    yield calls

def test_create_model(self):
        operation = migrations.CreateModel("Pony", [
            ('pony_id', models.AutoField(primary_key=True)),
            ('pink', models.IntegerField(default=1)),
        ])
        project_state = ProjectState()
        new_state = project_state.clone()
        operation.state_forwards('tests', new_state)
        self.assertTableNotExists('tests_pony')
        with connection.schema_editor() as editor:
            operation.database_forwards('tests', editor, project_state, new_state)
        self.assertTableExists('tests_pony')
        with connection.schema_editor() as editor:
            operation.database_backwards('tests', editor, new_state, project_state)
        self.assertTableNotExists('tests_pony')

def alter_db_table(field, filters: List[str]):
    """Creates a model with the specified field
    and then renames the database table.

    Arguments:
        field:
            The field to include into the
            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=[
                    ('title', field.clone())
                ]
            )
        ], project)

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

    yield calls

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 remove_field(field, filters: List[str]):
    """Removes the specified field from a model.

    Arguments:
        field:
            The field to remove from a model.

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

    model = define_fake_model({'title': field})
    project = migrations.state.ProjectState.from_apps(apps)

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

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

    yield calls

def rename_field(field, filters: List[str]):
    """Renames a field from one name to the other.

    Arguments:
        field:
            Field to be renamed.

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

    model = define_fake_model({'title': field})
    project = migrations.state.ProjectState.from_apps(apps)

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

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

    yield calls

def detect_soft_applied(self, project_state, migration):
        """
        Tests whether a migration has been implicitly applied - that the
        tables it would create exist. This is intended only for use
        on initial migrations (as it only looks for CreateModel).
        """
        # Bail if the migration isn't the first one in its app
        if [name for app, name in migration.dependencies if app == migration.app_label]:
            return False, project_state
        if project_state is None:
            after_state = self.loader.project_state((migration.app_label, migration.name), at_end=True)
        else:
            after_state = migration.mutate_state(project_state)
        apps = after_state.apps
        found_create_migration = False
        # Make sure all create model are done
        for operation in migration.operations:
            if isinstance(operation, migrations.CreateModel):
                model = apps.get_model(migration.app_label, operation.name)
                if model._meta.swapped:
                    # We have to fetch the model to test with from the
                    # main app cache, as it's not a direct dependency.
                    model = global_apps.get_model(model._meta.swapped)
                if model._meta.db_table not in self.connection.introspection.table_names(self.connection.cursor()):
                    return False, project_state
                found_create_migration = True
        # If we get this far and we found at least one CreateModel migration,
        # the migration is considered implicitly applied.
        return found_create_migration, after_state

def reduce_model_create_delete(self, operation, other, in_between):
        """
        Folds a CreateModel and a DeleteModel into nothing.
        """
        if (operation.name_lower == other.name_lower and
                not operation.options.get("proxy", False)):
            return []

def reduce_model_create_rename(self, operation, other, in_between):
        """
        Folds a model rename into its create
        """
        if operation.name_lower == other.old_name_lower:
            return [
                migrations.CreateModel(
                    other.new_name,
                    fields=operation.fields,
                    options=operation.options,
                    bases=operation.bases,
                    managers=operation.managers,
                )
            ]

def reduce_create_model_alter_field(self, operation, other, in_between):
        if operation.name_lower == other.model_name_lower:
            return [
                migrations.CreateModel(
                    operation.name,
                    fields=[
                        (n, other.field if n == other.name else v)
                        for n, v in operation.fields
                    ],
                    options=operation.options,
                    bases=operation.bases,
                    managers=operation.managers,
                )
            ]

def reduce_create_model_remove_field(self, operation, other, in_between):
        if operation.name_lower == other.model_name_lower:
            return [
                migrations.CreateModel(
                    operation.name,
                    fields=[
                        (n, v)
                        for n, v in operation.fields
                        if n.lower() != other.name_lower
                    ],
                    options=operation.options,
                    bases=operation.bases,
                    managers=operation.managers,
                )
            ]

def test_migrations():
    """Tests whether the migrations are properly generated and executed."""

    simulator = MigrationSimulator()

    Model = simulator.define_model(
        fields={
            'id': models.IntegerField(primary_key=True),
            'name': models.CharField(max_length=255, null=True),
            'other_name': models.CharField(max_length=255)
        },
        meta_options={
            'indexes': [
                ConditionalUniqueIndex(
                    fields=['name', 'other_name'],
                    condition='"name" IS NOT NULL',
                    name='index1'
                ),
                ConditionalUniqueIndex(
                    fields=['other_name'],
                    condition='"name" IS NULL',
                    name='index2'
                )
            ]
        }
    )

    migration = simulator.make_migrations()
    assert len(migration.operations) == 3

    operations = migration.operations
    assert isinstance(operations[0], CreateModel)

    for operation in operations[1:]:
        assert isinstance(operation, AddIndex)

    calls = [call[0] for _, call, _ in simulator.migrate('CREATE UNIQUE INDEX')[0]['CREATE UNIQUE INDEX']]

    db_table = Model._meta.db_table
    assert calls[0] == 'CREATE UNIQUE INDEX "index1" ON "{0}" ("name", "other_name") WHERE "name" IS NOT NULL'.format(
        db_table
    )
    assert calls[1] == 'CREATE UNIQUE INDEX "index2" ON "{0}" ("other_name") WHERE "name" IS NULL'.format(
        db_table
    )

    with transaction.atomic():
        Model.objects.create(id=1, name="name", other_name="other_name")
        with pytest.raises(IntegrityError):
            Model.objects.create(id=2, name="name", other_name="other_name")

    with transaction.atomic():
        Model.objects.create(id=1, name=None, other_name="other_name")
        with pytest.raises(IntegrityError):
            Model.objects.create(id=2, name=None, other_name="other_name")