Python 现代开发学习笔记

(二十一)管理命令导入书目与维护权限

seed_shelf 读取前面采集的 CSV,准备 alice、bob 和后台维护账号。命令文件放进应用的 management/commands,Django 根据文件名提供子命令。

# 项目文件:reading_project/src/library/management/commands/seed_shelf.py
# 先准备账户与权限组,再从 CSV 导入书籍;重复执行时复用已经存在的对象。
import csv
import json
from pathlib import Path

from django.contrib.auth import get_user_model
from django.contrib.auth.models import Group, Permission
from django.core.management.base import BaseCommand
from django.db import transaction

from library.models import Book, BookChange, Tag


class Command(BaseCommand):
    help = "导入采集书目和演示账号,保留已有密码与阅读状态"

    def add_arguments(self, parser):
        parser.add_argument("--csv", type=Path, default=Path("data/books.csv"))

    @transaction.atomic
    def handle(self, *args, **options):
        User = get_user_model()
        users = {}
        for name in ["alice", "bob", "manager"]:
            user, created = User.objects.get_or_create(username=name)
            if created:
                user.set_password(f"learn-{name}")
                user.is_staff = name == "manager"
                user.save()
            users[name] = user
        export = Permission.objects.get(
            content_type__app_label="library", codename="export_book"
        )
        users["alice"].user_permissions.add(export)
        group, _ = Group.objects.get_or_create(name="书架维护")
        group.permissions.set(
            Permission.objects.filter(
                content_type__app_label="library",
                codename__in=["add_book", "change_book", "delete_book", "view_book"],
            )
        )
        users["manager"].groups.add(group)
        with options["csv"].open(encoding="utf-8-sig", newline="") as source:
            for row in csv.DictReader(source):
                self.add_book(
                    users["alice"],
                    row["title"],
                    bool(int(row["finished"])),
                    json.loads(row["tags"]),
                )
        self.add_book(users["manager"], "Django 后台记录", False, ["工程"])
        self.stdout.write(self.style.SUCCESS("已准备账号、书目与维护权限。"))

BaseCommand.add_arguments 扩展命令行参数,handle 执行本次任务。get_user_model 取得当前用户模型,set_password 生成摘要。账号已经存在就保留密码,初始化不覆盖读者之后修改的内容。

Group 将一组模型权限一起授予 manager;alice 单独取得导出权限。权限已经由迁移后的处理流程建立,因此先 migrate 再执行这个命令。

# 续写同一文件:reading_project/src/library/management/commands/seed_shelf.py
    # 只有首次创建书籍时写入标签,重复导入会保留用户随后修改过的标签。
    @staticmethod
    def add_book(owner, title, finished, names):
        book, created = Book.objects.get_or_create(
            owner=owner, title=title, defaults={"finished": finished}
        )
        if created:
            book.tags.set([Tag.objects.get_or_create(name=name)[0] for name in names])
            BookChange.objects.create(book=book, action="created")

get_or_create 按用户与书名查重,只有新书才设置初始标签和修改记录。整个导入使用 atomic,中途出错不会留下半批书。

先迁移建立权限和表,再按账号查重与分配组,逐本按归属书名查重,只有新记录设置初始标签并提交

笔记正文与全部题解 · Ctrl / ⌘ K