116 lines
4.0 KiB
Python
116 lines
4.0 KiB
Python
# dao/students_dao.py
|
|
# 本文件封装对 model 表的所有数据库操作（增、删、改、查）
|
|
|
|
|
|
from sqlalchemy.orm import Session
|
|
from fastapi import HTTPException
|
|
from sqlalchemy_fastapi_demo_1.model.c_lass import Classinfo
|
|
from sqlalchemy_fastapi_demo_1.model.advisors import Advisor
|
|
from sqlalchemy_fastapi_demo_1.model.students import Student
|
|
from sqlalchemy_fastapi_demo_1.scheme.students import StudentCreate, StudentUpdate
|
|
from typing import Optional, List
|
|
|
|
|
|
class StudentDAO:
|
|
|
|
@staticmethod
|
|
def inspect_class_id_unq(db: Session, class_id: int):
|
|
"""校验班级是否存在：去 c_lass 表查"""
|
|
return db.query(Classinfo).filter(Classinfo.class_id == class_id).first()
|
|
|
|
@staticmethod
|
|
def inspect_advisor_id_unq(db: Session, advisor_id: int):
|
|
"""校验顾问是否存在：去 advisors 表查"""
|
|
return db.query(Advisor).filter(Advisor.advisor_id == advisor_id).first()
|
|
|
|
@staticmethod
|
|
def create_student(db: Session, obj_in: StudentCreate):
|
|
"""新增学生，检查学号唯一性"""
|
|
exist = db.query(Student).filter(Student.stu_id == obj_in.stu_id).first()
|
|
# 只按 stu_id 判断，且包含逻辑删除的记录
|
|
if exist:
|
|
if exist.is_deleted == 1:
|
|
raise HTTPException(status_code=400, detail="该学号已被占用（处于逻辑删除状态），请先恢复")
|
|
raise HTTPException(status_code=400, detail="该学生已存在")
|
|
db_obj = Student(**obj_in.model_dump())
|
|
try:
|
|
db.add(db_obj)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
except Exception:
|
|
db.rollback()
|
|
raise
|
|
return db_obj
|
|
|
|
@staticmethod
|
|
def get_students_all(db: Session, skip: int = 0, limit: int = 100):
|
|
"""分页查询全部学生"""
|
|
return db.query(Student).filter(Student.is_deleted == 0).offset(skip).limit(limit).all()
|
|
|
|
@staticmethod
|
|
def get_student_by_id(db: Session, stu_id: int):
|
|
"""根据学生编号单条查询"""
|
|
return db.query(Student).filter(Student.is_deleted == 0,
|
|
Student.stu_id == stu_id).first()
|
|
|
|
@staticmethod
|
|
def query_multi_condition(
|
|
db: Session,
|
|
stu_id: Optional[int] = None,
|
|
stu_name: Optional[str] = None,
|
|
class_id: Optional[int] = None,
|
|
skip: int = 0,
|
|
limit: int = 100
|
|
) :
|
|
q = db.query(Student).filter(Student.is_deleted == 0)
|
|
"""多条件选择性查询"""
|
|
# 只有参数不为None的时候，才加上查询条件
|
|
if stu_id is not None:
|
|
q = q.filter(Student.stu_id == stu_id)
|
|
if stu_name is not None:
|
|
q = q.filter(Student.stu_name.like(f"%{stu_name}%"))
|
|
if class_id is not None:
|
|
q = q.filter(Student.class_id == class_id)
|
|
|
|
q = q.offset(skip).limit(limit)
|
|
return q.all()
|
|
|
|
@staticmethod
|
|
def update_student(db: Session,stu_id:int, obj: StudentUpdate):
|
|
db_obj = db.query(Student).filter(Student.stu_id == stu_id,
|
|
Student.is_deleted == 0).first()
|
|
if not db_obj:
|
|
raise HTTPException(status_code=404, detail="学生数据不存在或已删除")
|
|
|
|
update_data = obj.model_dump(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(db_obj, field, value)
|
|
try:
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
except Exception:
|
|
db.rollback()
|
|
raise
|
|
return db_obj
|
|
|
|
@staticmethod
|
|
def delete_student(db: Session, stu_id: int):
|
|
db_obj = db.query(Student).filter(Student.stu_id == stu_id,
|
|
Student.is_deleted == 0).first()
|
|
if not db_obj:
|
|
raise HTTPException(status_code=404, detail="学生数据不存在或已删除")
|
|
|
|
db_obj.is_deleted = 1
|
|
try:
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
except Exception:
|
|
db.rollback()
|
|
raise
|
|
return True # 返回布尔值，而不是 dict
|
|
|
|
|
|
|
|
|
|
|