126 lines
5.3 KiB
Python
126 lines
5.3 KiB
Python
from fastapi import Query,Depends
|
|
from sqlalchemy import func, case
|
|
from model.student_info import Student
|
|
from model.employments import Employments
|
|
from model.scores import Scores
|
|
from sqlalchemy.orm import Session
|
|
|
|
#student_info学生基本信息
|
|
#动态年龄范围查询：支持用户输入年龄阈值及比较条件（如大于、小于、等于、区间等），动态查询符合条件的学员信息
|
|
def student_info_age(db: Session,min_age:int ,max_age:int ,operator:str,):
|
|
if operator not in("大于","大于等于","小于","小于等于","区间"):
|
|
raise ValueError("输入比较格式不正确，请在规定范围内选择输入")
|
|
else:
|
|
q = db.query(Student)
|
|
if operator == "大于":
|
|
q = q.filter(Student.age > min_age)
|
|
elif operator == "大于等于":
|
|
q = q.filter(Student.age >= min_age)
|
|
elif operator == "小于":
|
|
q = q.filter(Student.age < min_age)
|
|
elif operator == "小于等于":
|
|
q = q.filter(Student.age <= min_age)
|
|
else: # 区间
|
|
q = q.filter(Student.age >= min_age, Student.age <= max_age)
|
|
print(q.all())
|
|
return q.all()
|
|
|
|
#多维度班级统计：统计每个班级的总人数，以及按性别（男、女）细分的人数分布
|
|
def student_info_class(db: Session):
|
|
s1 = (db.query(Student.class_name,
|
|
func.count(Student.id).label("总人数"),
|
|
func.sum(case((Student.gender=="男",1),else_=0)).label("男"),
|
|
func.sum(case((Student.gender=="女",1),else_=0)).label("女"))
|
|
.group_by(Student.class_name).all())
|
|
|
|
return [
|
|
{
|
|
"class_name": row.class_name,
|
|
"总人数": row.总人数,
|
|
"男": row.男,
|
|
"女": row.女,
|
|
}for row in s1
|
|
]
|
|
|
|
|
|
|
|
#student_score学生成绩
|
|
#查询每次考试成绩都在输入分数线（如80分）以上的学生的编号、姓名和成绩。
|
|
def student_score(db: Session,s_n : int):
|
|
s1 = db.query(Scores.id).group_by(Scores.id).having(func.min(Scores.score)>=s_n).all()
|
|
s1_ids = [row[0] for row in s1]
|
|
if not s1_ids:
|
|
return []
|
|
s2 = (db.query(Student.id,Student.name,Scores.exam_round,Scores.score)
|
|
.join(Student, Scores.id == Student.id)
|
|
.filter(Scores.id.in_(s1_ids)).all())
|
|
return [{"id": i.id, "name": i.name, "exam_round": i.exam_round, "score": i.score}
|
|
for i in s2]
|
|
#查询有输入指定次数（如两次(包含两次)）以上不及格的学生的姓名、班级和不及格成绩明细。
|
|
def student_score60(db: Session,n : int ):
|
|
s1 = (db.query(Scores.id).group_by(Scores.id)
|
|
.having(func.sum(case((Scores.score < 60, 1), else_=0)) >=n)
|
|
.all())
|
|
s1_ids = [row[0] for row in s1]
|
|
if not s1_ids:
|
|
return []
|
|
s2 = (db.query(Student.name,Student.class_name,Scores.exam_round,Scores.score)
|
|
.join(Student, Scores.id == Student.id)
|
|
.filter(Scores.id.in_(s1_ids)).all())
|
|
return [{"id": i.name, "name": i.class_name, "exam_round": i.exam_round, "score": i.score}
|
|
for i in s2]
|
|
#统计每次考试每个班级的平均分，并支持按分数从高到低或从低到高动态排序。
|
|
def student_score_class(db: Session,order: str = "desc"):
|
|
s1 = (db.query(Scores.exam_round,
|
|
Student.class_name,
|
|
func.avg(Scores.score).label("平均分"),).join(Student, Scores.id == Student.id)
|
|
.group_by(Student.class_name,Scores.exam_round))
|
|
if order == "desc":
|
|
res = s1.order_by(func.avg(Scores.score).desc()).all()
|
|
else:
|
|
res = s1.order_by(func.avg(Scores.score).asc()).all()
|
|
return [row._asdict() for row in res]
|
|
|
|
|
|
#student_employments
|
|
#查询查询就业薪资排名前n的学生
|
|
def student_employments_salary(db: Session,n:int):
|
|
s1 = db.query(Employments).order_by(Employments.sal.desc()).limit(n).all()
|
|
return s1
|
|
|
|
#统计每个学生的就业时长
|
|
def student_employments_time(db: Session):
|
|
time = []
|
|
all_ = db.query(Employments).all()
|
|
for employments in all_:
|
|
if employments.offer_review is None:
|
|
time.append({"name":employments.student.name,
|
|
"class_name": employments.student.class_name,
|
|
"status": "未就业"})
|
|
elif employments.offer_open and employments.offer_review:
|
|
time.append({"name":employments.student.name,
|
|
"class_name": employments.student.class_name,
|
|
"time": (employments.offer_review-employments.offer_open).days})
|
|
return time
|
|
|
|
#统计每个班级的平均就业时长（仅统计进入就业阶段，即有就业开放时间的学生）
|
|
def student_employments_class_time(db: Session):
|
|
s1 = (db.query(Student.class_name,
|
|
func.avg(
|
|
func.unix_timestamp(Employments.offer_review) -
|
|
func.unix_timestamp(Employments.offer_open))
|
|
.label("avg_time"))
|
|
.join(Student, Employments.id == Student.id)
|
|
.filter(Employments.offer_review.isnot(None))
|
|
.group_by(Student.class_name).all())
|
|
time = []
|
|
for i in s1:
|
|
days = round(float(i.avg_time) / 86400, 2) if i.avg_time else 0
|
|
time.append({
|
|
"class_name": i.class_name,
|
|
"平均时长（天）": days
|
|
})
|
|
return time
|
|
|
|
|