初始提交：学生管理系统，实现增删改查基础功能
This commit is contained in:
xy182
@@ -0,0 +1,428 @@
|
||||
|
||||
from sqlalchemy import func
|
||||
from sqlalchemy.exc import SQLAlchemyError
|
||||
from sqlalchemy.orm import Session
|
||||
from datetime import date
|
||||
|
||||
from model import Student, Classes, Score, Employment
|
||||
from schema.student_schema import AgeQuery
|
||||
|
||||
|
||||
class BusinessException(Exception):
|
||||
"""业务异常类，DAO层捕获数据库异常后抛出，由API层处理"""
|
||||
pass
|
||||
|
||||
|
||||
class DuplicateStudentNo(BusinessException):
|
||||
"""学号已存在。继承 BusinessException，API 层可以单独挑出来返回 409。"""
|
||||
pass
|
||||
|
||||
#=============把一个学生对象变成字典形式================
|
||||
def student_dict(student):
|
||||
if student is None:
|
||||
return {}
|
||||
return {"s_id": student.sid,
|
||||
"student_no": student.student_no,
|
||||
"student_name":student.student_name,
|
||||
"class_id":student.class_id,
|
||||
"flag":student.flag,
|
||||
"advisor_id":student.advisor_id,
|
||||
"native_place":student.native_place,
|
||||
"school":student.school,
|
||||
"major":student.major,
|
||||
"enrollment_time":student.enrollment_time,
|
||||
"graduation_time":student.graduation_time,
|
||||
"education":student.education,
|
||||
"age":student.age,
|
||||
"gender":student.gender,
|
||||
"state":student.state}
|
||||
|
||||
|
||||
def create_student(db: Session,
|
||||
student_no: str,
|
||||
student_name: str,
|
||||
class_id: int = None,
|
||||
flag: int = 1,
|
||||
advisor_id: int = None,
|
||||
native_place: str = None,
|
||||
school: str = None,
|
||||
major: str = None,
|
||||
enrollment_time: date = None,
|
||||
graduation_time: date = None,
|
||||
education: str = None,
|
||||
age: int = None,
|
||||
gender: str = None,
|
||||
state: str = None
|
||||
):
|
||||
"""
|
||||
创建学生
|
||||
"""
|
||||
try:
|
||||
# 学号是唯一约束，先查一下，给出比 IntegrityError 更好懂的提示
|
||||
existing = db.query(Student).filter(Student.student_no == student_no).first()
|
||||
if existing is not None:
|
||||
raise DuplicateStudentNo(f"学号 {student_no} 已存在")
|
||||
|
||||
new_student = Student(
|
||||
student_no=student_no,
|
||||
student_name=student_name,
|
||||
class_id=class_id,
|
||||
flag=flag,
|
||||
advisor_id=advisor_id,
|
||||
native_place=native_place,
|
||||
school=school,
|
||||
major=major,
|
||||
enrollment_time=enrollment_time,
|
||||
graduation_time=graduation_time,
|
||||
education=education,
|
||||
age=age,
|
||||
gender=gender,
|
||||
state=state,
|
||||
)
|
||||
db.add(new_student)
|
||||
db.commit()
|
||||
db.refresh(new_student)
|
||||
return new_student
|
||||
except SQLAlchemyError as e:
|
||||
db.rollback()
|
||||
raise BusinessException(f"创建学生失败: {str(e)}")
|
||||
|
||||
#================查询学生==================
|
||||
def search_student(db: Session, student_no: str,
|
||||
student_name: str,
|
||||
class_id: int = None,
|
||||
page: int = 1,
|
||||
size:int=5):
|
||||
try:
|
||||
query = db.query(Student).filter(Student.flag == 1)
|
||||
|
||||
if student_no is not None:
|
||||
query = query.filter(Student.student_no == student_no)
|
||||
if student_name:
|
||||
query = query.filter(Student.student_name.like(f"%{student_name}%"))
|
||||
if class_id is not None:
|
||||
query = query.filter(Student.class_id == class_id)
|
||||
|
||||
total = query.count()
|
||||
offset = (page - 1) * size
|
||||
students = query.order_by(Student.sid.desc()).offset(offset).limit(size).all()
|
||||
|
||||
items = [student_dict(s) for s in students]
|
||||
pages = (total + size - 1) // size if size > 0 else 0
|
||||
|
||||
return {
|
||||
"items": items,
|
||||
"total": total,
|
||||
"page": page,
|
||||
"page_size": size,
|
||||
"pages": pages,
|
||||
}
|
||||
except SQLAlchemyError as e:
|
||||
raise BusinessException(f"查询学生列表失败: {str(e)}")
|
||||
#=================ID查询学生详情====================
|
||||
def get_student(db: Session, sid: int):
|
||||
"""
|
||||
根据ID查询学生，自动过滤逻辑删除
|
||||
"""
|
||||
try:
|
||||
student = db.query(Student).filter(
|
||||
Student.sid == sid,
|
||||
Student.flag == 1
|
||||
).first()
|
||||
return student
|
||||
except SQLAlchemyError as e:
|
||||
raise BusinessException(f"查询学生失败: {str(e)}")
|
||||
|
||||
|
||||
#===============更新学生信息================
|
||||
def update_student(db: Session, sid: int,
|
||||
student_no: str = None,
|
||||
student_name: str = None,
|
||||
class_id: int = None,
|
||||
flag: int = None,
|
||||
advisor_id: int = None,
|
||||
native_place: str = None,
|
||||
school: str = None,
|
||||
major: str = None,
|
||||
enrollment_time: date = None,
|
||||
graduation_time: date = None,
|
||||
education: str = None,
|
||||
age: int = None,
|
||||
gender: str = None,
|
||||
state: str = None):
|
||||
"""
|
||||
更新学生信息：只改传了值的字段，没传（None）的保持原样。
|
||||
"""
|
||||
try:
|
||||
student = db.query(Student).filter(Student.sid == sid).first()
|
||||
if student is None:
|
||||
return None
|
||||
|
||||
# 判断的是"传进来的参数"，是否是None
|
||||
if student_no is not None:
|
||||
student.student_no = student_no
|
||||
if student_name is not None:
|
||||
student.student_name = student_name
|
||||
if class_id is not None:
|
||||
student.class_id = class_id
|
||||
if flag is not None:
|
||||
student.flag = flag
|
||||
if advisor_id is not None:
|
||||
student.advisor_id = advisor_id
|
||||
if native_place is not None:
|
||||
student.native_place = native_place
|
||||
if school is not None:
|
||||
student.school = school
|
||||
if major is not None:
|
||||
student.major = major
|
||||
if enrollment_time is not None:
|
||||
student.enrollment_time = enrollment_time
|
||||
if graduation_time is not None:
|
||||
student.graduation_time = graduation_time
|
||||
if education is not None:
|
||||
student.education = education
|
||||
if age is not None:
|
||||
student.age = age
|
||||
if gender is not None:
|
||||
student.gender = gender
|
||||
if state is not None:
|
||||
student.state = state
|
||||
|
||||
db.commit()
|
||||
db.refresh(student)
|
||||
return student
|
||||
except SQLAlchemyError as e:
|
||||
db.rollback()
|
||||
raise BusinessException(f"学生更新信息失败: {str(e)}")
|
||||
|
||||
#==============逻辑删除学生==================
|
||||
def delete_student(db: Session, sid: int):
|
||||
try:
|
||||
student = db.query(Student).filter(Student.sid == sid).first()
|
||||
if student is None or student.flag == 0:
|
||||
return None
|
||||
student.flag = 0
|
||||
db.commit() # 进行任务提交
|
||||
db.refresh(student)
|
||||
return student
|
||||
except SQLAlchemyError as e:
|
||||
db.rollback()
|
||||
raise BusinessException(f"学生删除失败: {str(e)}")
|
||||
|
||||
|
||||
#==============性别查询学生==================
|
||||
def get_class_gender(db: Session):
|
||||
class_list = db.query(Classes).all()
|
||||
res = []
|
||||
for cls in class_list:
|
||||
cid = cls.cid
|
||||
# 当前班级有效学生总数
|
||||
class_total = (
|
||||
db.query(func.count(Student.sid))
|
||||
.filter(Student.class_id == cid, Student.flag == 1)
|
||||
.scalar()
|
||||
)
|
||||
|
||||
# 当前班级男生数量
|
||||
male_count = (
|
||||
db.query(func.count(Student.sid))
|
||||
.filter(Student.class_id == cid, Student.flag == 1, Student.gender == "男")
|
||||
.scalar()
|
||||
)
|
||||
|
||||
# 当前班级女生数量
|
||||
female_count = (
|
||||
db.query(func.count(Student.sid))
|
||||
.filter(Student.class_id == cid, Student.flag == 1, Student.gender == "女")
|
||||
.scalar()
|
||||
)
|
||||
|
||||
res.append({
|
||||
"class_name": cls.class_name,
|
||||
"class_total": class_total,
|
||||
"boys_count": male_count,
|
||||
"girls_count": female_count
|
||||
})
|
||||
return res
|
||||
|
||||
#==============查询每次考试高于自定义分数线的学生==================
|
||||
def get_higher_score(db:Session,score):
|
||||
all_students = (db.query(Student)
|
||||
.filter(Student.flag==1)
|
||||
.all())
|
||||
higherStu=[]
|
||||
for student in all_students:
|
||||
all_scores = []
|
||||
for stu_score in student.scores:
|
||||
if stu_score.score<score:
|
||||
break
|
||||
all_scores.append(stu_score.score)
|
||||
else:
|
||||
if len(all_scores) > 0:
|
||||
higherStu.append({
|
||||
'stu_id':student.sid,
|
||||
'stu_name':student.student_name,
|
||||
'stu_score':all_scores
|
||||
})
|
||||
return higherStu
|
||||
|
||||
#==============查询不及格次数大于指定次数的学生==================
|
||||
def get_fail_more_than(db: Session, fail_times: int):
|
||||
# 查询有效学生
|
||||
all_students = (db.query(Student)
|
||||
.filter(Student.flag == 1)
|
||||
.all())
|
||||
result = []
|
||||
for student in all_students:
|
||||
#不及格成绩列表
|
||||
fail_score_list = []
|
||||
for stu_score in student.scores:
|
||||
if stu_score.score < 60:
|
||||
fail_score_list.append(stu_score.score)
|
||||
if len(fail_score_list) >= fail_times:
|
||||
result.append({
|
||||
'stu_id': student.sid,
|
||||
'stu_name': student.student_name,
|
||||
'class_name': student.classes.class_name,
|
||||
'fail_scores': fail_score_list
|
||||
})
|
||||
return result
|
||||
|
||||
|
||||
#==============查询所有学生就业时长==================
|
||||
def work_time(db:Session):
|
||||
all_times = []
|
||||
#所有工作信息
|
||||
worksInfo = db.query(Employment).filter(Employment.flag == 1).all()
|
||||
#所有学生信息
|
||||
all_students = (db.query(Student).filter(Student.flag==1).all())
|
||||
for work in worksInfo:
|
||||
if work.offer_time is not None and work.employment_start_time is not None and work.employment_start_time >= work.offer_time:
|
||||
delta = work.employment_start_time - work.offer_time
|
||||
days = delta.days # 整数天数
|
||||
for i in all_students:
|
||||
if i.sid==work.student_id:
|
||||
all_times.append({'name':i.student_name,'id': work.student_id, 'work_time': days})
|
||||
return all_times
|
||||
|
||||
#==============统计平均分,可以切换排序方式==================
|
||||
def count_avg(db: Session,order_type: str = "desc"):
|
||||
avg_score = func.avg(Score.score).label("avg_score")
|
||||
query = (
|
||||
db.query(
|
||||
Score.score.label("score"),
|
||||
Classes.cid.label("class_id"),
|
||||
Classes.class_name.label("class_name"),
|
||||
avg_score,
|
||||
)
|
||||
.join(Student, Score.student_id == Student.sid)
|
||||
.join(Classes, Student.class_id == Classes.cid)
|
||||
.group_by(Score.score, Classes.cid, Classes.class_name)
|
||||
)
|
||||
|
||||
# 动态排序
|
||||
if order_type == "asc":
|
||||
query = query.order_by(avg_score.asc())
|
||||
else:
|
||||
query = query.order_by(avg_score.desc())
|
||||
rows = query.all()
|
||||
# Row 转 dict，方便上层使用
|
||||
return [
|
||||
{
|
||||
"score": row.score,
|
||||
"class_id": row.class_id,
|
||||
"class_name": row.class_name,
|
||||
"avg_score": float(row.avg_score) if row.avg_score is not None else 0.0,
|
||||
}
|
||||
for row in rows
|
||||
]
|
||||
|
||||
#==============按年龄区间自定义比较==================
|
||||
def get_students_by_age(db: Session, query:AgeQuery):
|
||||
q = db.query(Student).filter(Student.flag==1)
|
||||
|
||||
op = query.operator
|
||||
if op == "gt":
|
||||
q = q.filter(Student.age > query.value)
|
||||
elif op == "lt":
|
||||
q = q.filter(Student.age < query.value)
|
||||
elif op == "eq":
|
||||
q = q.filter(Student.age == query.value)
|
||||
elif op == "ge":
|
||||
q = q.filter(Student.age >= query.value)
|
||||
elif op == "le":
|
||||
q = q.filter(Student.age <= query.value)
|
||||
elif op == "between":
|
||||
q = q.filter(Student.age.between(query.min_value, query.max_value))
|
||||
return q.all()
|
||||
|
||||
#==============按薪资排名==================
|
||||
def get_top_n_salary(db: Session, rank: int):
|
||||
rows = (
|
||||
db.query(
|
||||
Student.student_name.label("student_name"),
|
||||
Classes.class_name.label("class_name"),
|
||||
Employment.offer_time.label("offer_time"),
|
||||
Employment.company_name.label("company"),
|
||||
Employment.salary.label("salary"),
|
||||
)
|
||||
.join(Student, Employment.student_id == Student.sid)
|
||||
.join(Classes, Student.class_id == Classes.cid)
|
||||
.order_by(Employment.salary.desc())
|
||||
.limit(rank)
|
||||
.all()
|
||||
)
|
||||
|
||||
return [
|
||||
{
|
||||
"student_name": row.student_name,
|
||||
"class_name": row.class_name,
|
||||
"offer_time": row.offer_time,
|
||||
"company": row.company,
|
||||
"salary": float(row.salary) if row.salary is not None else 0.0,
|
||||
}
|
||||
for row in rows
|
||||
]
|
||||
|
||||
#==============每个班的平均就业时长==================
|
||||
def avg_work_time_by_class(db: Session):
|
||||
rows = (
|
||||
db.query(
|
||||
Classes.cid.label("class_id"),
|
||||
Classes.class_name.label("class_name"),
|
||||
Employment.employment_start_time,
|
||||
Employment.offer_time,
|
||||
)
|
||||
.join(Student, Employment.student_id == Student.sid)
|
||||
.join(Classes, Student.class_id == Classes.cid)
|
||||
.filter(Employment.employment_start_time.isnot(None))
|
||||
.filter(Employment.offer_time.isnot(None))
|
||||
.filter(Employment.flag == 1, Student.flag == 1)
|
||||
.filter(Employment.employment_start_time >= Employment.offer_time)
|
||||
.all()
|
||||
)
|
||||
|
||||
# 普通字典：{class_id: {"class_name": ..., "days_list": [...]}}
|
||||
class_data = {}
|
||||
|
||||
for row in rows:
|
||||
delta = row.employment_start_time - row.offer_time
|
||||
days = delta.total_seconds() / 86400
|
||||
if row.class_id not in class_data:
|
||||
class_data[row.class_id] = {
|
||||
"class_name": row.class_name,
|
||||
"days_list": [],
|
||||
}
|
||||
class_data[row.class_id]["days_list"].append(days)
|
||||
# 组装结果
|
||||
result = []
|
||||
for class_id, data in class_data.items():
|
||||
days_list = data["days_list"]
|
||||
avg_days = round(sum(days_list) / len(days_list), 2)
|
||||
result.append({
|
||||
"class_id": class_id,
|
||||
"class_name": data["class_name"],
|
||||
"days": avg_days,
|
||||
})
|
||||
|
||||
return result
|
||||
Reference in New Issue
Block a user