SST/PythonProject/dao/student_dao.py

from fastapi import Depends
from util.database import get_db
from model.all_model import Student_Model
from typing import List, Optional, Dict, Any
from datetime import datetime

def add_student_dao(o,db):
    try:
        o1 = Student_Model( **o)
        db.add(o1)
        db.commit()
        stu_id=o1.stu_id
        return db.query(Student_Model).filter(Student_Model.stu_id == stu_id).first()
    except:
        db.rollback()
        return None

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):
    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
                   ) -> tuple[List[Dict[str, Any]], int]:
    try:
        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 and stu_name.strip() != "":
          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,total
    except:
        db.rollback()
        return [],0