55 lines
1.9 KiB
Python
55 lines
1.9 KiB
Python
"""顾问 DAO（需求第 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 Advisor, Student
|
|
|
|
|
|
class AdvisorDao(BaseDao[Advisor]):
|
|
model = Advisor
|
|
|
|
@classmethod
|
|
def build_stmt(
|
|
cls,
|
|
keyword: str | None = None,
|
|
dept: str | None = None,
|
|
gender: int | None = None,
|
|
order_by: str = "id",
|
|
order: str = "desc",
|
|
) -> Select:
|
|
stmt = select(Advisor).where(Advisor.alive())
|
|
if keyword:
|
|
like = f"%{keyword.strip()}%"
|
|
stmt = stmt.where(
|
|
or_(Advisor.name.like(like), Advisor.advisor_no.like(like), Advisor.phone.like(like))
|
|
)
|
|
if dept:
|
|
stmt = stmt.where(Advisor.dept == dept)
|
|
if gender:
|
|
stmt = stmt.where(Advisor.gender == gender)
|
|
|
|
sortable = {"id": Advisor.id, "advisor_no": Advisor.advisor_no, "name": Advisor.name}
|
|
column = sortable.get(order_by or "id", Advisor.id)
|
|
stmt = stmt.order_by(column.desc() if (order or "desc").lower() == "desc" else column.asc())
|
|
return stmt
|
|
|
|
@classmethod
|
|
def next_advisor_no(cls, db: Session) -> str:
|
|
prefix = "A"
|
|
last = db.scalar(select(func.max(Advisor.advisor_no)).where(Advisor.advisor_no.like(f"{prefix}%")))
|
|
seq = int(last[len(prefix):]) + 1 if last and last[len(prefix):].isdigit() else 1
|
|
return f"{prefix}{seq:04d}"
|
|
|
|
@classmethod
|
|
def student_count_map(cls, db: Session) -> dict[int, int]:
|
|
stmt = (
|
|
select(Student.advisor_id, func.count(Student.id))
|
|
.where(Student.alive(), Student.advisor_id.is_not(None))
|
|
.group_by(Student.advisor_id)
|
|
)
|
|
return {row[0]: row[1] for row in db.execute(stmt).all()}
|