from sqlalchemy.orm import Session from model.stu_model import Student from sqlalchemy.exc import SQLAlchemyError from schema.stu_schema import StudentCreateRequest,StudentUpdateRequest def create_student(db: Session, student_create: StudentCreateRequest): try: db_student = Student(**student_create.model_dump()) db.add(db_student) db.commit() except SQLAlchemyError: db.rollback() return None new_student = db.query(Student).filter( Student.stu_id == student_create.stu_id, Student.is_deleted == False).first() return new_student def get_student_by_id(db: Session, stu_id: str): query = db.query(Student).filter( Student.stu_id == stu_id, Student.is_deleted == False) return query.first() def get_student_list(db: Session, stu_id: str | None = None, class_id: str = None, page: int = 1, page_size: int = 10, stu_name: str = None): page = max(1, page) page_size = max(1, min(100, page_size)) query = db.query(Student).filter(Student.is_deleted == False) if stu_id: query = query.filter(Student.stu_id == stu_id) if stu_name: query = query.filter(Student.stu_name.like(f"%{stu_name}%")) if class_id: query = query.filter(Student.class_id == class_id) total = query.count() offset = (page - 1) * page_size db_student_list = query.offset(offset).limit(page_size).all() return total, db_student_list def update_student(db: Session, stu_id: str, student_update: StudentUpdateRequest): try: db_student = get_student_by_id(db=db, stu_id=stu_id) if not db_student: return None update_data = student_update.model_dump(exclude_unset=True) for k, v in update_data.items(): setattr(db_student, k, v) db.commit() except SQLAlchemyError: db.rollback() return None return db_student def delete_student_logic(db: Session, stu_id: str): try: db_student = get_student_by_id(db, stu_id) if not db_student: return None db_student.is_deleted = True db.commit() except SQLAlchemyError: db.rollback() return None return db_student