xzbautomarkapi/admin/api/endpoints/school/schoolclass.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import copy
from typing import Optional, Union

from fastapi import APIRouter, Depends, Query, Path
from sqlalchemy import text, desc
from sqlalchemy.ext.asyncio import AsyncSession
from starlette.background import BackgroundTasks

from bgtask.tasks import bgtask_delete_related_object
from crud.school import crud_class, crud_school, crud_grade
from models.school import SchoolClass
from models.user import Admin
from schemas.school.school import (SchoolClassList, NewClassInfo, SchoolClassDetail, ClassInDB,
                                   UpdateClass)
from utils.depends import get_async_db, get_current_user

router = APIRouter()


# 班级列表
@router.get("/classes",
            response_model=SchoolClassList,
            response_model_exclude_none=True,
            summary="班级列表")
async def get_classes(page: Optional[int] = None,
                      size: Optional[int] = None,
                      sid: int = Query(None, description="学校ID"),
                      gid: int = Query(None, description="年级ID"),
                      db: AsyncSession = Depends(get_async_db),
                      current_user: Admin = Depends(get_current_user)):
    _q = {}
    if sid is not None:
        _q["school_id"] = sid
    if gid is not None:
        _q["grade_id"] = gid
    if ((page is not None) and page >= 1) and ((size is not None) and size >= 1):
        offset = (page - 1) * size
    else:
        offset = size = None

    total, items = await crud_class.find_all(db, filters=_q, offset=offset, limit=size)
    return {"total": total, "data": items}


@router.post("/classes",
             response_model=SchoolClassDetail,
             response_model_exclude_none=True,
             summary="新建班级")
async def create_class(info: NewClassInfo,
                       db: AsyncSession = Depends(get_async_db),
                       current_user: Admin = Depends(get_current_user)):
    info_dict = info.dict(exclude_none=True)
    if not info_dict:
        return {"errcode": 400, "mess": "提交参数为空!"}
    if not (("name" not in info_dict) or ("amount" not in info_dict)):    # 单个建班或快速建班
        return {"errcode": 400, "mess": "缺少班级名称或数量"}

    # 判断学校是否存在
    db_school = await crud_school.find_one(db,
                                           filters={"id": info_dict["school_id"]},
                                           return_fields=["name"])
    if not db_school:
        return {"errcode": 404, "mess": "学校不存在!"}
    else:
        info_dict["school_name"] = db_school.name

    # 判断年级是否存在
    db_grade = await crud_grade.find_one(db,
                                         filters={
                                             "school_id": info.school_id,
                                             "id": info.grade_id
                                         },
                                         return_fields=["name"])
    if not db_grade:
        return {"errcode": 404, "mess": "年级不存在!"}
    else:
        info_dict["grade_name"] = db_grade.name

    # 新建班级，如果存在amount，表示为快速建班，否则为普通建班
    new_classes = []
    info_dict["creator_id"] = info_dict["editor_id"] = current_user.id
    info_dict["creator_name"] = info_dict["editor_name"] = current_user.username
    if info_dict.get("amount", 0):
        # 查询当前学校+当前年级下是否存在班级
        _, db_class = await crud_class.find_all(db,
                                                filters={
                                                    "school_id": info.school_id,
                                                    "grade_id": info.grade_id
                                                },
                                                return_fields=["name"])
        db_class_names = [x.name for x in db_class]
        for i in range(1, info_dict.pop("amount") + 1):
            name = f"{info_dict['grade_name']}{str(i)}班"
            if name not in db_class_names:
                temp = copy.deepcopy(info_dict)
                temp["name"] = name
                new_classes.append(temp)
    else:
        info_dict["name"] = info_dict["grade_name"] + info_dict["name"]
        new_classes.append(ClassInDB(**info_dict))

    item = await crud_class.insert_many(db, new_classes)
    if item is None:
        return {"errcode": 400, "mess": "存在同名班级"}
    else:
        return {"data": item}


@router.get("/classes/{cid}",
            response_model=SchoolClassDetail,
            response_model_exclude_none=True,
            summary="班级详情")
async def get_class(cid: str = Path(..., description="班级ID"),
                    db: AsyncSession = Depends(get_async_db),
                    current_user: Admin = Depends(get_current_user)):
    db_obj = await crud_class.find_one(db, filters={"id": cid})
    return {"data": db_obj} if db_obj else {"errcode": 404, "mess": "班级不存在!"}


# 更新班级
@router.put("/classes/{cid}",
            response_model=SchoolClassDetail,
            response_model_exclude_none=True,
            summary="更新班级")
async def update_class(info: UpdateClass,
                       cid: int = Path(..., description="班级ID"),
                       db: AsyncSession = Depends(get_async_db),
                       current_user: Admin = Depends(get_current_user)):
    # 判断提交参数
    info_dict = info.dict(exclude_none=True)
    if not info_dict:
        return {"errcode": 400, "mess": "提交参数为空!"}

    # 判断班级是否存在
    db_obj = await crud_class.find_one(db, filters={"id": cid})
    if not db_obj:
        return {"errcode": 404, "mess": "班级不存在!"}

    # 如果学校ID不等于班级的school_id, 则更新
    if ("school_id" in info_dict) and (db_obj.school_id != info_dict["school_id"]):
        db_school = await crud_school.count(db,
                                            filters={"id": info.school_id},
                                            return_fields=["name"])
        if not db_school:
            return {"errcode": 404, "mess": "学校不存在!"}
        else:
            info.school_name = db_school.name

    # 判断年级是否存在
    if ("grade_id" in info_dict) and (db_obj.grade_id != info_dict["grade_id"]):
        db_grade = await crud_grade.count(db,
                                          filters={
                                              "school_id": info.school_id,
                                              "id": info.grade_id
                                          },
                                          return_fields=["name"])
        if not db_grade:
            return {"errcode": 404, "mess": "年级不存在!"}
        else:
            info.grade_name = db_grade.name

    # 更新班级
    info.editor_id = current_user.id
    info.editor_name = current_user.username
    db_obj = await crud_class.update(db, db_obj, info)

    return {"data": db_obj}


# 删除班级
@router.delete("/classes/{cid}",
               response_model=SchoolClassDetail,
               response_model_exclude_none=True,
               summary="删除班级")
async def delete_class(bgtask: BackgroundTasks,
                       cid: Union[int,
                                  str] = Path(...,
                                              description="学生ID，批量删除传用逗号分隔ID(str)，单个删除传ID(int)"),
                       db: AsyncSession = Depends(get_async_db),
                       current_user: Admin = Depends(get_current_user)):
    # 查询所有班级
    if isinstance(cid, int):
        deleted_count = 1
        _q = [SchoolClass.id == cid]
    else:
        cids = [int(x.strip()) for x in cid.split(',') if x.strip()]
        deleted_count = len(cids)
        _q = [SchoolClass.id.in_(cids)]
    total, db_objs = await crud_class.find_all(db, filters=_q)
    # 删除
    for item in db_objs:
        await crud_class.delete(db, obj_id=item.id)
        bgtask.add_task(bgtask_delete_related_object, cid=item.id)
    return {
        "data": total,
        "mess": "success" if total == deleted_count else f"成功:{total}，失败: {deleted_count - total}"
    }