211 lines
6.6 KiB
Python
211 lines
6.6 KiB
Python
from typing import Any, List, Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Path, Response, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app import models, schemas
|
|
from app.api.v1.deps import get_db, get_current_active_user, get_admin_user
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("/", response_model=List[schemas.Course])
|
|
def read_courses(
|
|
db: Session = Depends(get_db),
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
teacher_id: Optional[int] = None,
|
|
current_user: models.User = Depends(get_current_active_user),
|
|
) -> Any:
|
|
"""
|
|
Retrieve courses.
|
|
"""
|
|
if teacher_id:
|
|
# Filter courses by teacher_id
|
|
courses = db.query(models.Course).filter(models.Course.teacher_id == teacher_id).offset(skip).limit(limit).all()
|
|
else:
|
|
# Get all courses
|
|
courses = db.query(models.Course).offset(skip).limit(limit).all()
|
|
|
|
return courses
|
|
|
|
|
|
@router.get("/{course_id}", response_model=schemas.CourseWithTeacher)
|
|
def read_course(
|
|
*,
|
|
db: Session = Depends(get_db),
|
|
course_id: int = Path(..., title="The ID of the course to get"),
|
|
current_user: models.User = Depends(get_current_active_user),
|
|
) -> Any:
|
|
"""
|
|
Get course by ID with teacher details.
|
|
"""
|
|
course = db.query(models.Course).filter(models.Course.id == course_id).first()
|
|
if not course:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="Course not found",
|
|
)
|
|
|
|
# Get teacher details if there is a teacher assigned
|
|
teacher = None
|
|
if course.teacher_id:
|
|
teacher = db.query(models.Teacher).filter(models.Teacher.id == course.teacher_id).first()
|
|
|
|
# Create a combined response
|
|
result = schemas.CourseWithTeacher.from_orm(course)
|
|
if teacher:
|
|
result.teacher = schemas.Teacher.from_orm(teacher)
|
|
|
|
return result
|
|
|
|
|
|
@router.post("/", response_model=schemas.Course)
|
|
def create_course(
|
|
*,
|
|
db: Session = Depends(get_db),
|
|
course_in: schemas.CourseCreate,
|
|
current_user: models.User = Depends(get_admin_user),
|
|
) -> Any:
|
|
"""
|
|
Create new course.
|
|
"""
|
|
# Check if course code is unique
|
|
existing_course = db.query(models.Course).filter(models.Course.course_code == course_in.course_code).first()
|
|
if existing_course:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Course with code {course_in.course_code} already exists",
|
|
)
|
|
|
|
# Check if teacher exists if provided
|
|
if course_in.teacher_id:
|
|
teacher = db.query(models.Teacher).filter(models.Teacher.id == course_in.teacher_id).first()
|
|
if not teacher:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"Teacher with ID {course_in.teacher_id} not found",
|
|
)
|
|
|
|
# Create the course
|
|
db_course = models.Course(
|
|
course_code=course_in.course_code,
|
|
title=course_in.title,
|
|
description=course_in.description,
|
|
credits=course_in.credits,
|
|
is_active=course_in.is_active,
|
|
teacher_id=course_in.teacher_id,
|
|
)
|
|
db.add(db_course)
|
|
db.commit()
|
|
db.refresh(db_course)
|
|
|
|
return db_course
|
|
|
|
|
|
@router.put("/{course_id}", response_model=schemas.Course)
|
|
def update_course(
|
|
*,
|
|
db: Session = Depends(get_db),
|
|
course_id: int = Path(..., title="The ID of the course to update"),
|
|
course_in: schemas.CourseUpdate,
|
|
current_user: models.User = Depends(get_admin_user),
|
|
) -> Any:
|
|
"""
|
|
Update a course.
|
|
"""
|
|
course = db.query(models.Course).filter(models.Course.id == course_id).first()
|
|
if not course:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="Course not found",
|
|
)
|
|
|
|
# Check if course code is unique if it's being updated
|
|
if course_in.course_code and course_in.course_code != course.course_code:
|
|
existing_course = db.query(models.Course).filter(models.Course.course_code == course_in.course_code).first()
|
|
if existing_course:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Course with code {course_in.course_code} already exists",
|
|
)
|
|
|
|
# Check if teacher exists if provided
|
|
if course_in.teacher_id and course_in.teacher_id != course.teacher_id:
|
|
teacher = db.query(models.Teacher).filter(models.Teacher.id == course_in.teacher_id).first()
|
|
if not teacher:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"Teacher with ID {course_in.teacher_id} not found",
|
|
)
|
|
|
|
# Update course fields
|
|
update_data = course_in.dict(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(course, field, value)
|
|
|
|
db.add(course)
|
|
db.commit()
|
|
db.refresh(course)
|
|
|
|
return course
|
|
|
|
|
|
@router.delete("/{course_id}", response_model=None, status_code=status.HTTP_204_NO_CONTENT)
|
|
def delete_course(
|
|
*,
|
|
db: Session = Depends(get_db),
|
|
course_id: int = Path(..., title="The ID of the course to delete"),
|
|
current_user: models.User = Depends(get_admin_user),
|
|
) -> Any:
|
|
"""
|
|
Delete a course.
|
|
"""
|
|
course = db.query(models.Course).filter(models.Course.id == course_id).first()
|
|
if not course:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="Course not found",
|
|
)
|
|
|
|
# Check if there are any exams associated with this course
|
|
exams = db.query(models.Exam).filter(models.Exam.course_id == course_id).first()
|
|
if exams:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Cannot delete course with associated exams",
|
|
)
|
|
|
|
# Check if there are any enrollments for this course
|
|
enrollments = db.query(models.ClassEnrollment).filter(models.ClassEnrollment.course_id == course_id).first()
|
|
if enrollments:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Cannot delete course with student enrollments",
|
|
)
|
|
|
|
db.delete(course)
|
|
db.commit()
|
|
|
|
return Response(status_code=status.HTTP_204_NO_CONTENT)
|
|
|
|
|
|
@router.get("/{course_id}/exams", response_model=List[schemas.Exam])
|
|
def read_course_exams(
|
|
*,
|
|
db: Session = Depends(get_db),
|
|
course_id: int = Path(..., title="The ID of the course to get exams for"),
|
|
current_user: models.User = Depends(get_current_active_user),
|
|
) -> Any:
|
|
"""
|
|
Get all exams for a course.
|
|
"""
|
|
course = db.query(models.Course).filter(models.Course.id == course_id).first()
|
|
if not course:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="Course not found",
|
|
)
|
|
|
|
exams = db.query(models.Exam).filter(models.Exam.course_id == course_id).all()
|
|
return exams |