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

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

import copy
from typing import Optional

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

from bgtask.tasks import delete_related_object
from crud.school import crud_class, crud_school, crud_grade
from models.user import SysUser
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: SysUser = Depends(get_current_user)):
    filters = {}
    if sid is not None:
        filters["school_id"] = sid
    if gid is not None:
        filters["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=filters,
                                             offset=offset,
                                             limit=size,
                                             order_by=[text("-created_at")])
    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: SysUser = 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, {"id": info_dict["school_id"]})
    if not db_school:
        return {"errcode": 404, "mess": "学校不存在!"}
    else:
        info_dict["school_name"] = db_school.name

    # 判断年级是否存在
    db_grade = await crud_grade.find_one(db, {
        "school_id": info.school_id,
        "id": info.grade_id
    })
    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: SysUser = Depends(get_current_user)):
    db_obj = await crud_class.find_one(db, {"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: SysUser = 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, {"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, {"id": info.school_id})
        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, {
            "school_id": info.school_id,
            "id": info.grade_id
        })
        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(bg_task: BackgroundTasks,
                       cid: int = Path(..., description="班级ID"),
                       db: AsyncSession = Depends(get_async_db),
                       current_user: SysUser = Depends(get_current_user)):
    existed = await crud_class.count(db, {"id": cid})
    if not existed:
        return {"errcode": 404, "mess": "班级不存在!"}
    else:
        await crud_class.delete(db, obj_id=cid)
        bg_task.add_task(delete_related_object, db, cid=cid)
        return {"data": None}