102 lines
3.4 KiB
Python
102 lines
3.4 KiB
Python
from pygments.lexers import sql
|
|
from sentry_sdk.integrations import sqlalchemy
|
|
|
|
from StatCalc.schema.statcalc_request import *
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import *
|
|
|
|
|
|
from students.model.students_model import Students
|
|
|
|
def get_age_30(db):
|
|
q = db.query(Students).filter(Students.age>=30).all()
|
|
return q
|
|
|
|
|
|
from class_management.model.class_management_model import ClassInfo
|
|
|
|
def get_class_count(g , db):
|
|
if g == "班级人数":
|
|
q = (db.query(ClassInfo.id,ClassInfo.name,func.count(ClassInfo.id).label('count') )
|
|
.join(Students,Students.class_id==ClassInfo.id)
|
|
.group_by(ClassInfo.id,ClassInfo.name,Students.sex)
|
|
.all())
|
|
elif g == "男女人数":
|
|
q = (db.query(Students.sex, func.count(Students.id).label('count'))
|
|
.group_by(Students.sex)
|
|
.all())
|
|
elif g == "班级中男女人数":
|
|
q = (db.query(ClassInfo.id,ClassInfo.name,Students.sex,func.count(ClassInfo.id).label('count') )
|
|
.join(Students,Students.class_id==ClassInfo.id)
|
|
.group_by(ClassInfo.id,ClassInfo.name,Students.sex)
|
|
.all())
|
|
else:
|
|
return False
|
|
return [row._asdict() for row in q]
|
|
|
|
|
|
from scores.model.score_model import Score
|
|
|
|
|
|
def get_excellent_students(db):
|
|
"""查询所有科目都在80分以上的优秀学生"""
|
|
# 先查总科目数
|
|
total_subjects = db.execute(text("SELECT COUNT(*) FROM subject")).scalar()
|
|
|
|
# 子查询：找出成绩>=80分的科目数等于总科目数的学生
|
|
sq = db.query(Students.id) \
|
|
.join(Score, Score.sid == Students.id) \
|
|
.filter(Score.score >= 80) \
|
|
.group_by(Students.id) \
|
|
.having(func.count(Score.id) == total_subjects) \
|
|
.subquery()
|
|
|
|
# 主查询：返回优秀学生信息
|
|
q = db.query(Students.id, Students.name, ClassInfo.name.label('class_name'), Score.score) \
|
|
.join(ClassInfo, ClassInfo.id == Students.class_id) \
|
|
.join(Score, Score.sid == Students.id) \
|
|
.filter(Students.id.in_(sq.select()), Score.score >= 80) \
|
|
.all()
|
|
return [row._asdict() for row in q]
|
|
|
|
|
|
from scores.model.score_model import Score
|
|
|
|
def get_score_80(db):
|
|
"""查询有两次以上不及格(<60分)的学生"""
|
|
# 子查询：找出不及格超过2门的学生ID
|
|
sq = db.query(Students.id) \
|
|
.join(Score, Score.sid == Students.id) \
|
|
.filter(Score.score < 60) \
|
|
.group_by(Students.id) \
|
|
.having(func.count(Score.id) > 2) \
|
|
.subquery()
|
|
|
|
# 主查询：返回不及格学生信息
|
|
q = db.query(Students.name, ClassInfo.name.label('class_name'), Score.score) \
|
|
.join(ClassInfo, ClassInfo.id == Students.class_id) \
|
|
.join(Score, Score.sid == Students.id) \
|
|
.filter(Students.id.in_(sq.select()), Score.score < 60) \
|
|
.all()
|
|
return [row._asdict() for row in q]
|
|
|
|
def get_class_avg_dao(db):
|
|
results = (db.query(Score.num,
|
|
Score.cid,
|
|
ClassInfo.name.label("class_name"),
|
|
func.avg(Score.score).label("avg_score"))
|
|
.join(ClassInfo, Score.cid == ClassInfo.id)
|
|
.filter(Score.is_deleted == 0)
|
|
.group_by(Score.num, Score.cid, ClassInfo.name)
|
|
.order_by(func.avg(Score.score).desc())
|
|
.all())
|
|
|
|
return [ {"num": r.num,
|
|
"cid": r.cid,
|
|
"class_name": r.class_name,
|
|
"avg_score": round(float(r.avg_score), 2),}
|
|
for r in results]
|
|
|
|
|
|
|