(二十一)管理命令导入书目与维护权限
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,中途出错不会留下半批书。