如何创建Django自定义迁移将JSONField数据迁移到独立模型字段

你的初始思路大方向没问题，但直接全量遍历调用save()、把加字段和数据填充、约束配置放在同阶段执行，很容易遇到约束冲突、性能拉胯、迁移中途失败回滚麻烦的问题，更稳妥的实现是拆成3个独立迁移分步执行，完全规避强约束适配的繁琐问题：

分步迁移流程 #

• 第一步：生成新增字段的初始迁移
  执行python manage.py makemigrations前，先给所有要新增的拆分字段临时配置null=True，不要加任何非空、唯一、外键级联这类强约束，生成迁移文件后检查确认字段配置，再执行该迁移。这一步的核心是先把字段结构加到数据库里，允许空值就不会因为存量数据无值触发约束报错。
• 第二步：编写自定义数据迁移填充值
  新建空迁移文件（执行python manage.py makemigrations --empty your_app_name），在迁移中编写数据填充逻辑，注意不要直接导入项目中定义的模型类，必须用apps.get_model()取迁移对应版本的历史模型，避免后续模型迭代修改后，历史迁移引用最新模型直接报错。
  填充逻辑参考如下：

  from django.db import migrations
  
  def fill_split_fields(apps, schema_editor):
      MyModel = apps.get_model("your_app_label", "YourModelName")
      batch_size = 1000
      offset = 0
      while True:
          # 分批取数避免大表一次性加载占满内存
          queryset = MyModel.objects.all()[offset:offset+batch_size]
          batch = list(queryset)
          if not batch:
              break
          to_update = []
          for instance in batch:
              raw_json = instance.old_json_blob or {}
              # 每个字段单独做取值、类型转换、缺省兜底，提前处理脏数据
              instance.new_field_a = raw_json.get("json_key_a", "")
              # 类型不兼容的场景加异常捕获，不要让单条脏数据打断整次迁移
              try:
                  instance.new_field_b = int(raw_json.get("json_key_b", 0))
              except (TypeError, ValueError):
                  instance.new_field_b = 0
              instance.new_field_c = raw_json.get("json_key_c") is True
              to_update.append(instance)
          # 用bulk_update批量更新，不触发模型save信号、不跑全量字段校验，性能远高于逐实例save
          MyModel.objects.bulk_update(
              to_update,
              ["new_field_a", "new_field_b", "new_field_c"],
              batch_size=batch_size
          )
          offset += batch_size
  
  def rollback_fill(apps, schema_editor):
      # 配套回滚逻辑，需要回滚时可以把新字段值重新写回原JSONField
      MyModel = apps.get_model("your_app_label", "YourModelName")
      batch_size = 1000
      offset = 0
      while True:
          queryset = MyModel.objects.all()[offset:offset+batch_size]
          batch = list(queryset)
          if not batch:
              break
          to_update = []
          for instance in batch:
              instance.old_json_blob = {
                  "json_key_a": instance.new_field_a,
                  "json_key_b": instance.new_field_b,
                  "json_key_c": instance.new_field_c
              }
              to_update.append(instance)
          MyModel.objects.bulk_update(
              to_update,
              ["old_json_blob"],
              batch_size=batch_size
          )
          offset += batch_size
  
  class Migration(migrations.Migration):
      dependencies = [
          # 替换为第一步新增字段的迁移文件标识，比如("your_app_label", "0013_add_split_fields")
      ]
  
      operations = [
          migrations.RunPython(fill_split_fields, rollback_fill),
      ]
  

• 第三步：加约束、清理旧字段
  等第二步的数据迁移执行完，所有存量数据都已经填充到新字段、脏数据也已经做了兜底处理，这时候再修改模型定义：给新字段加上你需要的非空、唯一、外键等约束，确认不需要保留原JSONField的话也可以标记为删除，再执行makemigrations生成第三个迁移。这时候加约束不会和存量数据冲突，不需要在数据填充阶段反复适配校验规则。

约束适配优化建议 #

• 永远不要把「加字段」「填数据」「加约束」三个动作合并到同一个迁移里，拆分后每个阶段职责单一，出问题只需要回滚对应阶段即可，不需要在数据填充的时候同时处理各种约束校验逻辑。
• 数据填充阶段不要调用逐实例的save()方法，bulk_update会跳过模型层的默认校验逻辑，你可以自己控制每个字段的清洗规则，避免无关的内置校验打断迁移。如果担心填充后的数据不符合约束，可以在迁移执行完后单独跑校验脚本筛出异常数据人工处理，不要把校验逻辑耦合在迁移流程里。
• 单表数据量超过10万的场景，把循环切片取数换成queryset.iterator(chunk_size=1000)，进一步降低内存占用，有条件的话可以给迁移加事务控制，避免长事务锁表影响线上业务。
• 线上执行迁移前，先拿生产库的脱敏备份在测试环境跑一遍全流程，统计脏数据的占比和类型，提前调整字段兜底规则，不要直接在线上环境调试迁移逻辑。

内容的提问来源于stack exchange，提问作者Iustinian Olaru