SST/PythonProject/dao/student_dao.py

from model.all_model import Student_Model
from typing import List, Optional, Dict, Any
from datetime import datetime
from sqlalchemy.exc import IntegrityError

def add_student_dao(o,db):
    if o.get('id_card'):
        conflict = (db.query(Student_Model)
                    .filter(Student_Model.id_card == o['id_card'],
                            Student_Model.delete_status == 0)
                    .first())
        if conflict:
            return 'conflict'
    try:
        o2 = Student_Model(**o)
        db.add(o2)
        db.commit()
        db.refresh(o2)  # 回填自增的 stu_id
        return o2
    except IntegrityError:
        db.rollback()
        return 'conflict'
    except Exception:
        db.rollback()
        return 'error'

def delete_student_dao(stu_id,db):
    try:
        rows = db.query(Student_Model).filter(Student_Model.stu_id == stu_id,Student_Model.delete_status == 0)\
            .update({'delete_status':1,'delete_time': datetime.now()})
        db.commit()
    except:
        db.rollback()
        rows = 0
    finally:
        return rows

def update_student_dao(stu_id,update_data,db):
    if update_data.get('id_card'):
        conflict = (db.query(Student_Model)
                    .filter(Student_Model.id_card == update_data['id_card'],
                            Student_Model.stu_id != stu_id,
                            Student_Model.delete_status == 0)
                    .first())
        if conflict:
            return 'conflict'

    try:
        rows = (db.query(Student_Model)
                .filter(Student_Model.stu_id == stu_id,
                        Student_Model.delete_status == 0)
                .update(update_data))
        db.commit()
    except IntegrityError:
        db.rollback()
        return 'conflict'
    except Exception:
        db.rollback()
        return 'error'

    return rows

def get_student_dao(stu_id:Optional[int]
                   ,stu_name:Optional[str]
                   ,class_id:Optional[int]
                   ,page: int
                   ,page_size: int
                   ,db
                   ) -> tuple[List[Dict[str, Any]], int]:
    try:
        q = db.query(Student_Model).filter(Student_Model.delete_status == 0)
        if stu_id:
          q = q.filter(Student_Model.stu_id == stu_id)
        if stu_name and stu_name.strip() != "":
          q = q.filter(Student_Model.stu_name.like(f"%{stu_name}%"))
        if class_id:
          q= q.filter(Student_Model.class_id == class_id)
        total = q.count()
        r = q.offset((page - 1) * page_size).limit(page_size).all()
        return r,total
    except:
        db.rollback()
        return [],0