Spaces:
Runtime error
Runtime error
File size: 3,463 Bytes
71a3948 93cf1dc 3358b33 93cf1dc 71a3948 93cf1dc 71a3948 93cf1dc 71a3948 93cf1dc 71a3948 93cf1dc 3358b33 93cf1dc 71a3948 93cf1dc 71a3948 93cf1dc 71a3948 93cf1dc 71a3948 93cf1dc 71a3948 93cf1dc 71a3948 93cf1dc 71a3948 93cf1dc 71a3948 93cf1dc |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 |
from sqlmodel import Session, select
from typing import List, Optional
from src.models import (
Student, StudentCreate, StudentUpdate, User, Role, ClearanceStatus, ClearanceDepartment, RFIDTag, UserCreate
)
from src.crud import users as user_crud
# --- Read Operations ---
def get_student_by_id(db: Session, student_id: int) -> Optional[Student]:
"""Retrieves a student by their primary key ID."""
return db.get(Student, student_id)
def get_student_by_matric_no(db: Session, matric_no: str) -> Optional[Student]:
"""Retrieves a student by their unique matriculation number."""
return db.exec(select(Student).where(Student.matric_no == matric_no)).first()
def get_student_by_tag_id(db: Session, tag_id: str) -> Optional[Student]:
"""Get student by RFID tag ID."""
from src.models import RFIDTag
tag = db.exec(select(RFIDTag).where(RFIDTag.tag_id == tag_id)).first()
if tag and tag.student_id:
return db.exec(select(Student).where(Student.id == tag.student_id)).first()
return None
def get_all_students(db: Session, skip: int = 0, limit: int = 100) -> List[Student]:
"""Retrieves a paginated list of all students."""
return db.exec(select(Student).offset(skip).limit(limit)).all()
# --- Write Operations ---
def create_student(db: Session, student_data: StudentCreate) -> Student:
"""
Creates a new student along with their associated user account for login
and initializes their clearance statuses.
"""
# 1. Create the associated User account for the student to log in
# The student's username is their matriculation number.
user_for_student = UserCreate(
username=student_data.matric_no,
password=student_data.password,
email=student_data.email,
full_name=student_data.full_name,
role=Role.STUDENT
)
# This might raise an exception if username/email exists, which is good.
db_user = user_crud.create_user(db, user=user_for_student)
# 2. Create the Student profile
db_student = Student(
full_name=student_data.full_name,
matric_no=student_data.matric_no,
email=student_data.email,
department=student_data.department,
# Note: The User linkage is handled via the RFID tag linking process
)
db.add(db_student)
db.commit()
db.refresh(db_student)
# 3. Initialize all clearance statuses for the new student
for dept in ClearanceDepartment:
status = ClearanceStatus(student_id=db_student.id, department=dept)
db.add(status)
db.commit()
db.refresh(db_student)
return db_student
def update_student(db: Session, student: Student, updates: StudentUpdate) -> Student:
"""Updates a student's profile information."""
update_data = updates.model_dump(exclude_unset=True)
student.sqlmodel_update(update_data)
db.add(student)
db.commit()
db.refresh(student)
return student
def delete_student(db: Session, student_id: int) -> Student | None:
"""Deletes a student and their associated clearance records."""
student_to_delete = db.get(Student, student_id)
if not student_to_delete:
return None
# Also delete the associated user account
user_to_delete = user_crud.get_user_by_username(db, username=student_to_delete.matric_no)
if user_to_delete:
db.delete(user_to_delete)
db.delete(student_to_delete)
db.commit()
return student_to_delete
|