Files
test/app/dao/teacher_dao.py
2026-09-21 19:03:31 +08:00

80 lines
2.8 KiB
Python

"""老师 DAO(需求 2.5)。"""
from __future__ import annotations
from sqlalchemy import Select, func, or_, select
from sqlalchemy.orm import Session
from app.dao.base_dao import BaseDao
from app.model import Clazz, Teacher, class_teachers
class TeacherDao(BaseDao[Teacher]):
model = Teacher
@classmethod
def build_stmt(
cls,
keyword: str | None = None,
subject: str | None = None,
title: str | None = None,
gender: int | None = None,
class_id: int | None = None,
order_by: str = "id",
order: str = "desc",
) -> Select:
stmt = select(Teacher).where(Teacher.alive())
if keyword:
like = f"%{keyword.strip()}%"
stmt = stmt.where(
or_(Teacher.name.like(like), Teacher.teacher_no.like(like), Teacher.phone.like(like))
)
if subject:
stmt = stmt.where(Teacher.subject.like(f"%{subject}%"))
if title:
stmt = stmt.where(Teacher.title == title)
if gender:
stmt = stmt.where(Teacher.gender == gender)
if class_id:
stmt = stmt.where(
Teacher.id.in_(
select(class_teachers.c.teacher_id).where(class_teachers.c.class_id == class_id)
)
)
sortable = {
"id": Teacher.id,
"teacher_no": Teacher.teacher_no,
"name": Teacher.name,
"hire_date": Teacher.hire_date,
"subject": Teacher.subject,
}
column = sortable.get(order_by or "id", Teacher.id)
stmt = stmt.order_by(column.desc() if (order or "desc").lower() == "desc" else column.asc())
return stmt
@classmethod
def get_by_no(cls, db: Session, teacher_no: str, with_deleted: bool = False) -> Teacher | None:
stmt = select(Teacher).where(Teacher.teacher_no == teacher_no)
if not with_deleted:
stmt = stmt.where(Teacher.alive())
return db.scalars(stmt).unique().first()
@classmethod
def next_teacher_no(cls, db: Session, year: int) -> str:
prefix = f"T{year}"
last = db.scalar(select(func.max(Teacher.teacher_no)).where(Teacher.teacher_no.like(f"{prefix}%")))
seq = int(last[len(prefix):]) + 1 if last and last[len(prefix):].isdigit() else 1
return f"{prefix}{seq:03d}"
@classmethod
def class_count_map(cls, db: Session) -> dict[int, int]:
stmt = (
select(class_teachers.c.teacher_id, func.count(class_teachers.c.class_id))
.select_from(class_teachers.join(Clazz, Clazz.id == class_teachers.c.class_id))
.where(Clazz.alive())
.group_by(class_teachers.c.teacher_id)
)
return {row[0]: row[1] for row in db.execute(stmt).all()}