70 lines
2.3 KiB
Python
70 lines
2.3 KiB
Python
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
|