能否在一次发布中完成Django多对多关系的Through模型切换？

原模型定义 #

class Book(models.Model):
    title = models.CharField(max_length=100)

class Author(models.Model):
    name = models.CharField(max_length=100)
    books = models.ManyToManyField(Book)

目标模型定义 #

需要切换为带自定义中间表的多对多关联，新增impact字段：

class Book(models.Model):
    title = models.CharField(max_length=100)

class BookAuthor(models.Model):
    book = models.ForeignKey(Book, on_delete=models.CASCADE)
    author = models.ForeignKey("Author", on_delete=models.CASCADE)
    impact = models.IntegerField(default=1)

    class Meta:
        unique_together = ("book", "author")

class Author(models.Model):
    name = models.CharField(max_length=100)
    books = models.ManyToManyField(Book, through=BookAuthor)

现有迁移脚本的问题 #

尝试编写的迁移脚本中，执行for book in author.books.all()时，会访问新创建的空BookAuthor表，而非原自动生成的多对多中间表，导致无法迁移原有关联数据：

from django.db import migrations

def migrate_author_books(apps, schema_editor):
    Author = apps.get_model('yourappname', 'Author')
    BookAuthor = apps.get_model('yourappname', 'BookAuthor')

    for author in Author.objects.all():
        for book in author.books.all():
            # Create a BookAuthor entry with default impact=1
            BookAuthor.objects.create(author=author, book=book, impact=1)

class Migration(migrations.Migration):
    dependencies = [
        ('yourappname', 'previous_migration_file'),
    ]
    operations = [
        migrations.CreateModel(name="BookAuthor", ...),
        migrations.RunPython(migrate_author_books),
        migrations.RemoveField(model_name="author", name="books"),
        migrations.AddField(model_name="author", name="books", field=models.ManyToManyField(...),
    ]

当前想到的分两次发布方案 #

• 添加新的BookAuthor模型并填充数据，保留原author.books字段，同时将代码中所有使用author.books的地方改为author.books_new
• 发布并执行迁移
• 删除原author.books字段，将books_new重命名为books，再次发布并执行迁移

更简便的解决方案 #

不需要分两次发布，只需调整迁移脚本，直接读取原自动生成的多对多中间表数据即可。具体修改如下：

调整后的迁移脚本 #

from django.db import migrations, connections

def migrate_author_books(apps, schema_editor):
    BookAuthor = apps.get_model('yourappname', 'BookAuthor')
    # 直接查询原自动生成的多对多中间表，表名格式为 {app名}_author_books
    with connections['default'].cursor() as cursor:
        cursor.execute("SELECT author_id, book_id FROM yourappname_author_books")
        rows = cursor.fetchall()
    
    # 批量创建BookAuthor记录，提升性能
    book_author_entries = [
        BookAuthor(author_id=row[0], book_id=row[1], impact=1)
        for row in rows
    ]
    BookAuthor.objects.bulk_create(book_author_entries)

class Migration(migrations.Migration):
    dependencies = [
        ('yourappname', 'previous_migration_file'),
    ]

    operations = [
        migrations.CreateModel(
            name='BookAuthor',
            fields=[
                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                ('impact', models.IntegerField(default=1)),
                ('book', models.ForeignKey(on_delete=models.CASCADE, to='yourappname.Book')),
                ('author', models.ForeignKey(on_delete=models.CASCADE, to='yourappname.Author')),
            ],
            options={
                'unique_together': {('book', 'author')},
            },
        ),
        migrations.RunPython(migrate_author_books),
        migrations.RemoveField(
            model_name='author',
            name='books',
        ),
        migrations.AddField(
            model_name='author',
            name='books',
            field=models.ManyToManyField(through='yourappname.BookAuthor', to='yourappname.Book'),
        ),
    ]

方案说明 #

1. 直接读取原中间表：在执行RunPython时，原自动生成的多对多中间表（yourappname_author_books）尚未被删除，通过原始SQL直接查询其中的author_id和book_id关联数据，避免了模型属性指向新中间表的问题。
2. 批量创建提升性能：使用bulk_create批量生成BookAuthor记录，比循环逐个创建更适合生产环境的大量数据场景。
3. 注意表名正确性：如果原多对多字段指定了db_table参数，需要将SQL中的表名替换为自定义的表名。

内容的提问来源于stack exchange，提问作者Martin Thoma