85 lines
2.8 KiB
Python
85 lines
2.8 KiB
Python
from fastapi import HTTPException,Depends
|
|
from util.database import get_db
|
|
from model.all_model import Student_Model
|
|
from typing import List, Optional, Dict, Any
|
|
from sqlalchemy import func
|
|
|
|
# 实现根据年龄查询学生信息的功能
|
|
def select_age(min_age,max_age,session=Depends(get_db)):
|
|
try:
|
|
if min_age is None and max_age is None:
|
|
raise HTTPException(status_code=400, detail="请至少传入一个参数")
|
|
session = session.query(Student_Model)
|
|
if min_age is not None:
|
|
session = session.filter(Student_Model.age >= min_age,Student_Model.delete_status == 0)
|
|
if max_age is not None:
|
|
session = session.filter(Student_Model.age <= max_age,Student_Model.delete_status == 0)
|
|
return session.all()
|
|
except HTTPException:
|
|
raise
|
|
except Exception:
|
|
raise HTTPException(status_code=500,detail="查询异常")
|
|
|
|
# 实现统计每个班级的学员总数以及男女生的总数的功能
|
|
def select_all(session=Depends(get_db)):
|
|
print('1234')
|
|
try:
|
|
l1 = session.query(Student_Model.class_id,func.count(1)).group_by(Student_Model.class_id,Student_Model.gender).all()
|
|
print(l1)
|
|
return l1
|
|
except Exception:
|
|
raise HTTPException(status_code=500,detail="查询异常")
|
|
|
|
def add_student_dao(o,db=Depends(get_db)):
|
|
try:
|
|
o1 = Student_Model( **o)
|
|
db.add(o1)
|
|
except:
|
|
db.rollback()
|
|
return False
|
|
else:
|
|
db.commit()
|
|
return True
|
|
|
|
def delete_student_dao(stu_id,db=Depends(get_db)):
|
|
try:
|
|
rows = db.query(Student_Model).filter(Student_Model.stu_id == stu_id,Student_Model.delete_status == 0)\
|
|
.update({'delete_status':1})
|
|
db.commit()
|
|
except:
|
|
db.rollback()
|
|
rows = 0
|
|
finally:
|
|
return rows
|
|
|
|
def update_student_dao(stu_id,update_data,db=Depends(get_db)):
|
|
try:
|
|
rows = db.query( Student_Model ).filter( Student_Model.stu_id == stu_id,Student_Model.delete_status == 0).update( update_data )
|
|
except:
|
|
db.rollback()
|
|
return False
|
|
else:
|
|
db.commit()
|
|
return rows
|
|
|
|
def get_student_dao(stu_id:Optional[int]
|
|
,stu_name:Optional[str]
|
|
,class_id:Optional[int]
|
|
,page: int
|
|
,page_size: int
|
|
,db=Depends(get_db)
|
|
) -> tuple[List[Dict[str, Any]], int]:
|
|
q = db.query(Student_Model)
|
|
if stu_id:
|
|
q = q.filter(Student_Model.stu_id == stu_id,Student_Model.delete_status == 0)
|
|
if stu_name:
|
|
q = q.filter(Student_Model.stu_name.like(f"%{stu_name}%"),Student_Model.delete_status == 0)
|
|
if class_id:
|
|
q= q.filter(Student_Model.class_id == class_id,Student_Model.delete_status == 0)
|
|
total = q.count()
|
|
r = q.offset((page - 1) * page_size).limit(page_size).all()
|
|
|
|
return r
|
|
|
|
|