[django]将json数据导入数据库的操纵过程

笑看天下无敌手  论坛元老 | 7 天前 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 2045|帖子 2045|积分 6135

马上注册，结交更多好友，享用更多功能，让你轻松玩转社区。

您需要 登录 才可以下载或查看，没有账号？立即注册

x
完整数据导入方案

  • 项目结构准备
    bash
    复制
    your_project/
    ├── your_app/
    │ ├── management/
    │ │ └── commands/
    │ │ └── import_members.py # 自定义管理命令
    │ ├── models.py
    │ └── fixtures/
    │ └── members.json # 生成的JSON数据
    ├── db.sqlite3
    └── manage.py
  • 模子定义（models.py）
    python
    复制
    from django.db import models
    from django.contrib.auth.models import AbstractBaseUser, PermissionsMixin
  1. class Member(AbstractBaseUser, PermissionsMixin):
  2.     username = models.CharField(max_length=255, unique=True)
  3.     password = models.CharField(max_length=255)
  4.     email = models.EmailField(max_length=50, unique=True, blank=True, null=True)
  5.     phone = models.CharField(max_length=50, unique=True, blank=True, null=True)
  6.     address = models.CharField(max_length=50, blank=True, null=True)
  7.     realname = models.CharField(max_length=50, blank=True, null=True)
  8.     nickname = models.CharField(max_length=50, blank=True, null=True)
  9.     wx_openid = models.CharField(max_length=50, blank=True, null=True)
  10.     avatar = models.CharField(max_length=100, blank=True, null=True)
  11.     coordinate = models.CharField(max_length=50, blank=True, null=True)
  12.     gender = models.CharField(max_length=50, blank=True, null=True)
  13.     birthday = models.DateField(null=True, blank=True)
  14.     mark = models.CharField(max_length=100, blank=True, null=True)
  15.     is_active = models.BooleanField(default=True)
  16.     is_staff = models.BooleanField(default=False)
  17.     created_at = models.DateTimeField(auto_now_add=True)
  18.     updated_at = models.DateTimeField(auto_now=True)
  20.     USERNAME_FIELD = 'username'
  21.     REQUIRED_FIELDS = []
复制代码

  • 自定义管理命令（import_members.py）
  1. from django.core.management.base import BaseCommand
  2. from django.contrib.auth.hashers import make_password
  3. from your_app.models import Member
  4. import json
  5. from datetime import datetime
  6. from django.db import transaction
  8. class Command(BaseCommand):
  9.     help = 'Import members from JSON file'
  11.     def add_arguments(self, parser):
  12.         parser.add_argument('file_path', type=str, help='Path to JSON file')
  14.     @transaction.atomic
  15.     def handle(self, *args, **options):
  16.         file_path = options['file_path']
  17.         processed = 0
  18.         skipped = 0
  19.         
  20.         try:
  21.             with open(file_path, 'r', encoding='utf-8') as f:
  22.                 data = json.load(f)
  23.                
  24.                 for item in data:
  25.                     # 处理日期字段
  26.                     try:
  27.                         birthday = datetime.strptime(item['birthday'], '%Y-%m-%d').date() if item['birthday'] else None
  28.                     except:
  29.                         birthday = None
  30.                         
  31.                     # 检查唯一约束
  32.                     if Member.objects.filter(username=item['username']).exists():
  33.                         self.stdout.write(self.style.WARNING(
  34.                             f"跳过重复用户: {item['username']}"
  35.                         ))
  36.                         skipped += 1
  37.                         continue
  39.                     # 创建记录
  40.                     Member.objects.create(
  41.                         username=item['username'],
  42.                         password=make_password(item['password'])
  43. ,
  44.                         phone=item['phone'],
  45.                         email=item['email'] or None,  # 空字符串转为None
  46.                         address=item['address'],
  47.                         realname=item['realname'],
  48.                         nickname=item['nickname'],
  49.                         wx_openid=item['wx_openid'],
  50.                         gender=item['gender'],
  51.                         birthday=birthday,
  52.                         mark=item['mark'],
  53.                         is_active=item['is_active'],
  54.                         is_staff=item['is_staff'],
  55.                         created_at=datetime.fromisoformat(item['created_at'].replace('Z', '')),
  56.                         # updated_at自动处理
  57.                     )
  58.                     processed += 1
  60.             self.stdout.write(self.style.SUCCESS(
  61.                 f"成功导入 {processed} 条记录，跳过 {skipped} 条重复记录"
  62.             ))
  64.         except Exception as e:
  65.             self.stdout.write(self.style.ERROR(f"导入失败: {str(e)}"))
复制代码

  • 操纵流程
    (1) 生成JSON数据
    利用之前改进的Python脚本生成members.json
(2) 放置数据文件
将生成的JSON文件放到项目目录：
bash
复制
cp members.json your_project/your_app/fixtures/
(3) 运行迁移
bash
复制
python manage.py makemigrations
python manage.py migrate
(4) 执行导入命令
bash
复制
python manage.py import_members your_app/fixtures/members.json
5. 关键增强功能
事件处置惩罚：
利用@transaction.atomic装饰器确保数据完整性
导入失败时主动回滚
数据清洗：
python
复制
处置惩罚空值

email=item[‘email’] or None # 将空字符串转为NULL
处置惩罚日期

birthday=datetime.strptime(…).date() if item[‘birthday’] else None
暗码加密：
  1. password=make_password(item['password'])
复制代码
唯一性检查：
  1. if Member.objects.filter(username=item['username']).exists():
  2.     continue
复制代码
错误处置惩罚：
捕捉全部异常并输堕落误信息
详细的日志输出

  • 高级配置建议
    批量创建优化：
  1. # 使用bulk_create提升性能
  2. batch_size = 100
  3. objs = []
  4. for item in data:
  5.     objs.append(Member(...))
  6.     if len(objs) >= batch_size:
  7.         Member.objects.bulk_create(objs)
  8.         objs = []
  9. if objs:
  10.     Member.objects.bulk_create(objs)
  11.     ```
  12. 进度显示：
  14. ```python
  15. from tqdm import tqdm
  16. for item in tqdm(data, desc='处理进度'):
  17.     # 处理逻辑
复制代码
邮件关照：
  1. from django.core.mail import send_mail
复制代码
在导入完成后发送关照邮件

日志记载：
  1. import logging
  2. logger = logging.getLogger(__name__)
  3. try:
  4.     # 导入逻辑
  5. except Exception as e:
  6.     logger.error(f"导入失败: {str(e)}")
  7.     ```
  8. 7. 验证导入结果
  9. 使用Django shell检查数据：
  11. ```bash
  12. python manage.py shell
复制代码
  1. from your_app.models import Member
  2. # 检查记录数量
  3. print(Member.objects.count())
  5. # 检查示例记录
  6. member = Member.objects.first()
  7. print(member.username, member.realname)
复制代码
检查暗码加密状态：
  1. print(member.password)  # 应该显示加密后的哈希值
复制代码

  • 常见问题处置惩罚
    日期格式错误：
在JSON生成阶段确保日期格式为ISO格式
添加更详细的日期解析错误提示
唯一约束冲突：
利用update_or_create方法处置惩罚重复数据
  1. Member.objects.update_or_create(
  2.     username=item['username'],
  3.     defaults={...}
  4. )
复制代码
内存优化：
利用ijson库处置惩罚大文件
  1. import ijson
  2. for item in ijson.items(f, 'item'):
复制代码
逐条处置惩罚

该方案提供了从数据准备到数据库导入的完整流程，包含错误处置惩罚、性能优化和安全措施，可直接用于生产环境。根据实际需求调解批量处置惩罚大小、日志级别和关照机制即可。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

笑看天下无敌手

论坛元老
这个人很懒什么都没写!

楼主热帖

标签云

渠道 国产数据库 集成商 AI 运维 CIO 存储 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表