如何在模型已编辑的Django应用SQLite数据中查找损坏的FK与M2M关联模型？

一、直接查询SQLite数据库定位损坏外键 #

SQLite默认可能未启用外键约束，先做基础检查再定位问题：

• 检查外键约束状态：

  PRAGMA foreign_keys;
  

  返回0则手动开启：

  PRAGMA foreign_keys = ON;
  

• 列出所有外键关联关系：

  SELECT 
      m.name AS 表名, 
      p.name AS 外键字段, 
      ref.master AS 关联表名, 
      ref.key AS 关联字段
  FROM sqlite_master m
  JOIN pragma_foreign_key_list(m.name) p
  JOIN pragma_index_list(ref.master) ref
  WHERE m.type = 'table';
  

• 针对单个外键字段查询损坏记录：
  比如article表的author_id关联author表的id，执行：

  SELECT id, author_id FROM article 
  WHERE author_id NOT IN (SELECT id FROM author);
  

二、用Django Shell批量检查模型外键 #

确保项目settings.py的数据库配置指向这份SQLite转储文件，然后通过shell批量校验：

• 启动Django Shell：

  python manage.py shell
  

• 执行批量检查脚本：

  from django.apps import apps
  
  for model in apps.get_models():
      print(f"检查模型: {model.__name__}")
      # 遍历所有外键字段
      for field in model._meta.get_fields():
          if field.is_relation and field.many_to_one:
              fk_field = field.name + "_id"
              # 获取所有非空的外键值
              all_fk_values = model.objects.values_list(fk_field, flat=True).distinct()
              existing_fks = [val for val in all_fk_values if val is not None]
              if not existing_fks:
                  continue
              # 关联模型的有效ID集合
              related_model = field.related_model
              valid_ids = set(related_model.objects.values_list('id', flat=True))
              # 筛选无效外键值
              invalid_fks = [val for val in existing_fks if val not in valid_ids]
              if invalid_fks:
                  print(f"  损坏外键字段 {fk_field}: 无效值 {invalid_fks}")
                  print(f"  涉及 {model.objects.filter(**{fk_field + '__in': invalid_fks}).count()} 条记录")
  

三、排查多对多关联损坏情况 #

多对多关系存储在中间表，需校验中间表记录是否在两个关联模型中都有对应ID：

• 在Django Shell中执行以下脚本：

  from django.apps import apps
  
  for model in apps.get_models():
      for field in model._meta.get_fields():
          if field.many_to_many:
              m2m_table = field.m2m_db_table()
              from_col = field.m2m_column_name()
              to_col = field.m2m_reverse_name()
              from_model = model
              to_model = field.related_model
              print(f"检查M2M关联: {from_model.__name__} <-> {to_model.__name__} (中间表: {m2m_table})")
              
              # 检查无效的关联源记录
              invalid_from_ids = [obj.id for obj in from_model.objects.raw(f"""
                  SELECT DISTINCT {from_col} FROM {m2m_table}
                  WHERE {from_col} NOT IN (SELECT id FROM {from_model._meta.db_table})
              """)]
              if invalid_from_ids:
                  print(f"  无效的{from_model.__name__} ID: {invalid_from_ids}")
              
              # 检查无效的关联目标记录
              invalid_to_ids = [obj.id for obj in to_model.objects.raw(f"""
                  SELECT DISTINCT {to_col} FROM {m2m_table}
                  WHERE {to_col} NOT IN (SELECT id FROM {to_model._meta.db_table})
              """)]
              if invalid_to_ids:
                  print(f"  无效的{to_model.__name__} ID: {invalid_to_ids}")
  

四、编写脚本生成占位符数据修复 #

确定损坏关联后，编写脚本批量生成占位符并修复：

1. 生成占位符实例工具函数：

   def create_placeholder(model, placeholder_data):
       # 按模型字段生成占位符记录，避免重复创建
       placeholder, _ = model.objects.get_or_create(**placeholder_data)
       return placeholder
   

2. 替换损坏外键：

   # 示例：为Article模型的无效author_id生成占位符作者
   placeholder_author = create_placeholder(Author, {"name": "占位符作者", "email": "placeholder@example.com"})
   # 批量更新损坏记录
   Article.objects.filter(author_id__in=invalid_fks).update(author=placeholder_author)
   

3. 清理无效多对多记录：

   from django.db import connection
   with connection.cursor() as cursor:
       cursor.execute(f"""
           DELETE FROM {m2m_table}
           WHERE {from_col} NOT IN (SELECT id FROM {from_model._meta.db_table})
           OR {to_col} NOT IN (SELECT id FROM {to_model._meta.db_table})
       """)
   # 可选：为缺失关联生成占位符后重新建立关联
   

内容的提问来源于stack exchange，提问作者Stuart Axon