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()
        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()
        return rows
    except Exception:
        db.rollback()
        raise

def update_student_dao(stu_id, update_data, db):
    if not update_data:
        return 0
    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]:
    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