123 lines
4.8 KiB
Python
123 lines
4.8 KiB
Python
"""老师管理接口（需求 2.5）。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Annotated
|
|
|
|
from fastapi import APIRouter, Query
|
|
|
|
from app.core.deps import DbSession, ReadAccount, WriteAccount
|
|
from app.core.exceptions import BusinessError, ConflictError, NotFoundError
|
|
from app.core.response import ok
|
|
from app.core.response import Resp
|
|
from app.core.utils import parse_date
|
|
from app.dao.clazz_dao import ClazzDao
|
|
from app.dao.teacher_dao import TeacherDao
|
|
from app.model import Teacher
|
|
from app.schema.common import PageResult
|
|
from app.schema.teacher_schema import TeacherCreate, TeacherOut, TeacherUpdate
|
|
|
|
router = APIRouter(prefix="/teachers", tags=["2.5 老师管理"])
|
|
|
|
|
|
@router.get("", response_model=Resp[PageResult[TeacherOut]], summary="老师列表（含带班信息）")
|
|
def list_teachers(
|
|
db: DbSession,
|
|
_: ReadAccount,
|
|
page: Annotated[int, Query(ge=1)] = 1,
|
|
page_size: Annotated[int, Query(ge=1, le=200)] = 10,
|
|
keyword: Annotated[str | None, Query(description="姓名/工号/电话")] = None,
|
|
subject: Annotated[str | None, Query(description="授课方向")] = None,
|
|
title: Annotated[str | None, Query(description="职称")] = None,
|
|
gender: Annotated[int | None, Query(ge=1, le=2)] = None,
|
|
class_id: Annotated[int | None, Query(description="带某个班的老师")] = None,
|
|
order_by: Annotated[str, Query()] = "id",
|
|
order: Annotated[str, Query(pattern="^(asc|desc)$")] = "desc",
|
|
):
|
|
stmt = TeacherDao.build_stmt(
|
|
keyword=keyword, subject=subject, title=title, gender=gender,
|
|
class_id=class_id, order_by=order_by, order=order,
|
|
)
|
|
items, total, page, pages = TeacherDao.paginate(db, stmt, page, page_size)
|
|
return ok({
|
|
"total": total, "page": page, "page_size": page_size, "pages": pages,
|
|
"items": [TeacherOut.model_validate(t) for t in items],
|
|
})
|
|
|
|
|
|
@router.post("", response_model=Resp[TeacherOut], summary="新增老师")
|
|
def create_teacher(db: DbSession, _: WriteAccount, payload: TeacherCreate):
|
|
teacher_no = (payload.teacher_no or "").strip()
|
|
if not teacher_no:
|
|
from datetime import date
|
|
|
|
year = (parse_date(payload.hire_date, "入职时间") or date.today()).year
|
|
teacher_no = TeacherDao.next_teacher_no(db, year)
|
|
if TeacherDao.get_by_no(db, teacher_no, with_deleted=True):
|
|
raise ConflictError(f"工号 {teacher_no} 已存在")
|
|
|
|
teacher = Teacher(
|
|
teacher_no=teacher_no,
|
|
name=payload.name,
|
|
gender=payload.gender,
|
|
phone=payload.phone,
|
|
email=payload.email,
|
|
title=payload.title,
|
|
subject=payload.subject,
|
|
hire_date=parse_date(payload.hire_date, "入职时间"),
|
|
remark=payload.remark,
|
|
)
|
|
if payload.class_ids:
|
|
teacher.classes = _resolve_classes(db, payload.class_ids)
|
|
|
|
db.add(teacher)
|
|
db.commit()
|
|
db.refresh(teacher)
|
|
return ok(TeacherOut.model_validate(teacher), msg=f"老师 {teacher.name} 已加入，工号 {teacher.teacher_no}")
|
|
|
|
|
|
@router.get("/{teacher_id}", response_model=Resp[TeacherOut], summary="老师详情")
|
|
def get_teacher(db: DbSession, _: ReadAccount, teacher_id: int):
|
|
return ok(TeacherOut.model_validate(TeacherDao.get_or_404(db, teacher_id, "老师")))
|
|
|
|
|
|
@router.put("/{teacher_id}", response_model=Resp[TeacherOut], summary="更新老师（含带班关系）")
|
|
def update_teacher(db: DbSession, _: WriteAccount, teacher_id: int, payload: TeacherUpdate):
|
|
teacher = TeacherDao.get_or_404(db, teacher_id, "老师")
|
|
data = payload.model_dump(exclude_unset=True)
|
|
if "hire_date" in data:
|
|
teacher.hire_date = parse_date(data.pop("hire_date"), "入职时间")
|
|
class_ids = data.pop("class_ids", None)
|
|
|
|
for key, value in data.items():
|
|
if value is not None and hasattr(teacher, key):
|
|
setattr(teacher, key, value)
|
|
|
|
if class_ids is not None:
|
|
teacher.classes = _resolve_classes(db, class_ids)
|
|
|
|
db.commit()
|
|
db.refresh(teacher)
|
|
return ok(TeacherOut.model_validate(teacher), msg="更新成功")
|
|
|
|
|
|
@router.delete("/{teacher_id}", summary="逻辑删除老师（仍带班时拒绝）")
|
|
def delete_teacher(db: DbSession, _: WriteAccount, teacher_id: int):
|
|
teacher = TeacherDao.get_or_404(db, teacher_id, "老师")
|
|
if teacher.class_names:
|
|
raise BusinessError(
|
|
f"老师「{teacher.name}」还在带 {len(teacher.class_names)} 个班："
|
|
f"{'、'.join(teacher.class_names)}。请先解除带班关系"
|
|
)
|
|
teacher.soft_delete()
|
|
db.commit()
|
|
return ok(msg=f"已删除老师 {teacher.name}")
|
|
|
|
|
|
def _resolve_classes(db, class_ids: list[int]) -> list:
|
|
all_classes = {c.id: c for c in ClazzDao.all(db, ClazzDao.build_stmt())}
|
|
missing = [cid for cid in class_ids if cid not in all_classes]
|
|
if missing:
|
|
raise NotFoundError(f"班级不存在：{missing}")
|
|
return [all_classes[cid] for cid in class_ids]
|