Django 数据库迁移:分阶段补齐唯一字段,并保住已有关系

数据库里已有数据时,修改模型不只是生成一条 DDL。新增唯一字段要为每条旧记录生成不同的值,多数据库项目要明确操作落在哪个连接,把多对多关系改成显式中间模型时还必须保住原来的关联表。本文按 Django 官方《How to create database migrations》6.1 版完整章节编译,说明这些迁移的组织方式。

版本与操作边界:本文核对的是 URL 中标明的 Django 6.1 文档,不据此推定读者项目版本。迁移会改变数据库结构或数据;文中代码仅作静态审查,未连接数据库、未执行迁移。应用前应按项目实际版本、数据库后端和表规模演练锁等待、并发写入、备份恢复与回滚。

Django 唯一字段迁移分成新增可空字段、逐行或分批回填、校验后施加唯一非空约束三个阶段;下方提示处理并发写入与数据库路由。
图 1:先扩展结构,再补齐数据,最后收紧约束。未完纪原创技术示意图,不是数据库运行截图。

一、多数据库:先判断连接,再把查询绑定到它

RunPython 回调接收历史应用注册表 apps 和 schema_editor。当前迁移使用的数据库别名在 schema_editor.connection.alias 中。如果迁移仅适用于默认库,可以直接判断:

from django.db import migrations

def forwards(apps, schema_editor):
    if schema_editor.connection.alias != "default":
        return
    # 在这里执行只针对 default 的数据操作。

class Migration(migrations.Migration):
    dependencies = []  # 按项目填写真实依赖。
    operations = [migrations.RunPython(forwards)]

另一种方法是交给数据库路由器。RunPython 的 hints 会作为 **hints 传给路由器的 allow_migrate():

class MyRouter:
    def allow_migrate(self, db, app_label, model_name=None, **hints):
        if "target_db" in hints:
            return db == hints["target_db"]
        return True

# 在 Migration.operations 中:
migrations.RunPython(forwards, hints={"target_db": "default"})

若一个 RunPython 或 RunSQL 只涉及一个模型,原文建议同时传入 model_name,尤其是可复用应用和第三方应用。这里的 return True 是原文的宽泛教学策略;多数据库项目应结合其他路由器审查它,不能把“未指定提示就允许所有库”当成通用配置。

编者补充:检查连接别名或提供路由提示,并不会自动把回调内所有 ORM 查询切换到该连接。下文改编代码显式使用 .using(db_alias)、save(using=db_alias) 和 transaction.atomic(using=db_alias),避免迁移连接与数据查询连接分离。

二、向已有记录的表增加唯一字段

假设模型需要一个默认值为 uuid.uuid4、不可为空且唯一的 UUIDField。直接给已有记录的表添加这样的字段可能失败:用于补齐旧行的默认值只生成一次,旧行得到相同值,违反唯一约束。正确的组织方式是把“新增字段”“回填数据”“建立约束”拆开。

  1. 在模型中定义最终字段:uuid = models.UUIDField(default=uuid.uuid4, unique=True)。这里传入的是函数,不是提前求值的 uuid.uuid4()。
  2. 运行 makemigrations,获得带 AddField 的迁移;再对同一应用运行两次 makemigrations myapp --empty,生成两个空迁移。下列文件名只是有意义的示例,真实依赖必须以项目为准。
  3. 把自动生成的字段操作复制到最后一个迁移,将 AddField 改为 AlterField;在第一个迁移中去掉 unique=True,改用 null=True。
  4. 在中间迁移中给每条旧记录生成 UUID。确认数据满足条件后,最后的迁移再施加唯一、非空约束。
# 0004_add_uuid_field.py
import uuid
from django.db import migrations, models

class Migration(migrations.Migration):
    dependencies = [("myapp", "0003_auto_20150129_1705")]
    operations = [
        migrations.AddField(
            model_name="mymodel",
            name="uuid",
            field=models.UUIDField(default=uuid.uuid4, null=True),
        ),
    ]

第一个迁移仍保留原文的 default。因此不能简单假定旧行全部为 NULL:在这个原文方案中,中间迁移要遍历所有旧行重新赋值。

# 0005_populate_uuid_values.py
# 与原文相比:显式绑定数据库别名,并用 iterator 减少一次装载的对象数。
import uuid
from django.db import migrations

def gen_uuid(apps, schema_editor):
    MyModel = apps.get_model("myapp", "MyModel")
    db_alias = schema_editor.connection.alias
    for row in MyModel.objects.using(db_alias).all().iterator():
        row.uuid = uuid.uuid4()
        row.save(using=db_alias, update_fields=["uuid"])

class Migration(migrations.Migration):
    dependencies = [("myapp", "0004_add_uuid_field")]
    operations = [
        migrations.RunPython(gen_uuid, reverse_code=migrations.RunPython.noop),
    ]
# 0006_remove_uuid_null.py
import uuid
from django.db import migrations, models

class Migration(migrations.Migration):
    dependencies = [("myapp", "0005_populate_uuid_values")]
    operations = [
        migrations.AlterField(
            model_name="mymodel",
            name="uuid",
            field=models.UUIDField(default=uuid.uuid4, unique=True),
        ),
    ]

之后才能按常规使用 migrate 应用迁移。RunPython.noop 只表示反向迁移不执行数据操作,不表示可以恢复回填之前的值。若不希望此步可逆,可以不提供反向函数;真正的数据回滚仍需要另行设计。

原文明确提醒有并发竞态:如果在 AddField 与 RunPython 之间仍有新对象写入,遍历全部记录会覆盖这些对象最初的 UUID。若外部系统已经保存了该 UUID,影响可能超出本表。停写窗口、分阶段发布或可识别的待回填标记应在演练中确定,不能把这三个文件视作天然零停机方案。

三、大表回填:非原子迁移与分批事务

在支持 DDL 事务的数据库上,例如 PostgreSQL 和 SQLite,迁移默认运行于事务内。为避免大表数据迁移形成一个过长事务,可以在迁移类上设置 atomic = False,然后只把每批更新放入事务。也可以在非原子迁移的某个 RunPython 上指定 atomic=True。

下面保留原文每批 1000 行、仅处理空 UUID 的思路,并补上数据库别名。它是另一种回填形态,前提是待处理的行确实以 NULL 标识;不能不加调整就替换前一节保留默认值的迁移。

import uuid
from django.db import migrations, transaction

def gen_uuid(apps, schema_editor):
    MyModel = apps.get_model("myapp", "MyModel")
    db_alias = schema_editor.connection.alias
    pending = MyModel.objects.using(db_alias).filter(uuid__isnull=True)
    while pending.exists():
        with transaction.atomic(using=db_alias):
            for row in pending.order_by("pk")[:1000]:
                row.uuid = uuid.uuid4()
                row.save(using=db_alias, update_fields=["uuid"])

class Migration(migrations.Migration):
    atomic = False
    # 仍需补齐项目中的真实 dependencies。
    operations = [migrations.RunPython(gen_uuid)]

order_by("pk") 是本稿为批次可追踪性增加的排序,并不提供并发锁。这个循环也不解决并发任务争抢、持续插入空值或其他进程改写 UUID 的问题。非原子迁移失败时,之前提交的批次会保留,重试策略必须考虑部分完成状态。

对于不支持 DDL 事务的后端,例如 MySQL、Oracle,迁移类上的 atomic 不会带来相同效果。原文特别区分了 MySQL 的“原子 DDL”:它针对单条语句,不等于多条 DDL 可以包在一个可整体回滚的事务中。

四、用依赖图控制顺序

Django 不按迁移文件名的字典序决定执行顺序,而是根据 dependencies 与 run_before 构建依赖图。通常用 dependencies 声明前置迁移:

class Migration(migrations.Migration):
    dependencies = [("myapp", "0123_the_previous_migration")]

若必须让第三方应用的迁移在自己的迁移之后执行,例如替换 AUTH_USER_MODEL 后安排其他应用,可以使用:

run_before = [("third_party_app", "0001_do_awesome")]

能改后续迁移的 dependencies 时优先使用它;只有无法或不适合修改后续迁移时才用 run_before。不要只重命名文件来期望改变执行顺序。

五、在第三方应用之间搬运数据

从旧应用搬到新应用时,需要同时照顾“旧应用仍安装”和“全新部署已没有旧应用”两种状态。原文的做法是在迁移依赖中有条件地加入旧应用,并在读取其历史模型时捕获 LookupError。数据回调使用传入的历史 apps,全局应用注册表只用来判断旧应用是否安装。

from django.apps import apps as global_apps
from django.db import migrations

def forwards(apps, schema_editor):
    try:
        OldModel = apps.get_model("old_app", "OldModel")
    except LookupError:
        return
    NewModel = apps.get_model("new_app", "NewModel")
    db_alias = schema_editor.connection.alias
    NewModel.objects.using(db_alias).bulk_create(
        NewModel(new_attribute=old_object.old_attribute)
        for old_object in OldModel.objects.using(db_alias).all()
    )

class Migration(migrations.Migration):
    dependencies = [
        ("myapp", "0123_the_previous_migration"),
        ("new_app", "0001_initial"),
    ]
    if global_apps.is_installed("old_app"):
        dependencies.append(("old_app", "0001_initial"))
    operations = [migrations.RunPython(forwards, migrations.RunPython.noop)]

这里仍只是一对一字段映射示例。bulk_create() 不等于自动去重或幂等迁移;大数据量、目标表唯一约束、重试时重复记录和对象批次内存都要另行处理。原文允许反向迁移不做任何事,也允许按实际业务删除新应用里的部分或全部迁入数据;后一种方案有破坏性,不能在没有数据归属标记时直接照搬。

六、把多对多字段改为显式 through 模型

已有 Book.authors 多对多关系时,直接改成使用 AuthorBook 中间模型,自动生成的迁移可能删除旧关联表、再创建新表,导致原有关联丢失。原文用 SeparateDatabaseAndState 把数据库操作和 Django 记录的模型状态分开:数据库只重命名旧表,状态层则登记中间模型并调整多对多字段。

应先通过 sqlmigrate 或 dbshell 查清现有表名和约束名,用新中间模型的 _meta.db_table 确认目标表名;外键字段名也要与 Django 原来的中间表一致。原文的关键结构如下,模型状态完整保留在示例中:

from django.db import migrations, models
import django.db.models.deletion

class Migration(migrations.Migration):
    dependencies = [("core", "0001_initial")]
    operations = [
        migrations.SeparateDatabaseAndState(
            database_operations=[
                migrations.RunSQL(
                    sql="ALTER TABLE core_book_authors RENAME TO core_authorbook",
                    reverse_sql="ALTER TABLE core_authorbook RENAME TO core_book_authors",
                ),
            ],
            state_operations=[
                migrations.CreateModel(
                    name="AuthorBook",
                    fields=[
                        ("id", models.AutoField(auto_created=True,
                            primary_key=True, serialize=False, verbose_name="ID")),
                        ("author", models.ForeignKey(
                            on_delete=django.db.models.deletion.DO_NOTHING,
                            to="core.Author")),
                        ("book", models.ForeignKey(
                            on_delete=django.db.models.deletion.DO_NOTHING,
                            to="core.Book")),
                    ],
                    options={"constraints": [
                        models.UniqueConstraint(fields=["author", "book"],
                                                name="unique_author_book"),
                    ]},
                ),
                migrations.AlterField(
                    model_name="book", name="authors",
                    field=models.ManyToManyField(to="core.Author",
                                                through="core.AuthorBook"),
                ),
            ],
        ),
        migrations.AddField(
            model_name="authorbook", name="is_primary",
            field=models.BooleanField(default=False),
        ),
    ]

新增的 is_primary 要放在 SeparateDatabaseAndState 之后。状态层的 CreateModel 并没有再次创建真实表;因此已有主键类型、外键删除行为和唯一约束必须与声明一致。特别是示例使用 AutoField,不能未经核对替代真实项目的主键类型。SQL 中两个表名是已核验名称的固定示例,不应拼接来自请求的标识符;本稿没有执行这些重命名语句。

七、把 unmanaged 模型改成 managed

如果模型原先设了 managed=False,应先移除该设置并单独生成迁移,再进行其他结构变化。原文指出,与 Meta.managed 切换放在同一迁移中的结构变更可能不会被应用。将两步分开,能让迁移状态与数据库管理责任的交接更加明确。

来源、署名与审阅说明

原文:Django 6.1 — How to create database migrations。作者:Django 文档贡献者;© 2005–2026 Django Software Foundation and individual contributors。本文由未完纪于 2026-10-05 中文编译并补充数据库别名、批次重试及 SQL 风险说明;相关代码按 Django BSD 三条款许可证保留署名,许可证全文见随稿来源文件。

本文覆盖源页全部实质章节,合并了重复导入与操作说明;所有编者改动均已标注。仅完成静态审查,未进行数据库执行测试,没有发现的问题不等于不存在漏洞。

保留的完整许可声明

以下为原项目适用许可文本,原作者与文档归属及本文改动说明见正文。

Copyright (c) Django Software Foundation and individual contributors.
All rights reserved.

Redistribution and use in source and binary forms, with or without modification,
are permitted provided that the following conditions are met:

    1. Redistributions of source code must retain the above copyright notice,
       this list of conditions and the following disclaimer.
    2. Redistributions in binary form must reproduce the above copyright
       notice, this list of conditions and the following disclaimer in the
       documentation and/or other materials provided with the distribution.

    3. Neither the name of Django nor the names of its contributors may be used
       to endorse or promote products derived from this software without
       specific prior written permission.

THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
© 版权声明
THE END
喜欢就支持一下吧
点赞0 分享
评论 抢沙发

请登录后发表评论

    暂无评论内容