284 lines
10 KiB
Python
284 lines
10 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, UploadFile, File
|
|
from sqlalchemy import select, func
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from app.core.deps import (
|
|
ensure_class_permission,
|
|
get_effective_class_permissions,
|
|
require_role,
|
|
resolve_class_id_for_user,
|
|
)
|
|
from app.db.database import get_db
|
|
from app.db.models import User, Class_
|
|
from app.schemas.assignment import (
|
|
AssignmentCreate, AssignmentUpdate, AssignmentOut,
|
|
AssignmentDetailOut, SubmissionGrade, SubmissionOut,
|
|
)
|
|
from app.schemas.common import PageResponse
|
|
from app.services.assignment_service import (
|
|
create_assignment,
|
|
update_assignment,
|
|
delete_assignment,
|
|
get_assignment_by_id,
|
|
list_assignments,
|
|
add_attachments,
|
|
create_submission,
|
|
get_submission_by_student,
|
|
grade_submission,
|
|
list_submissions,
|
|
)
|
|
from app.services.cos_service import upload_file
|
|
|
|
router = APIRouter(prefix="/api/assignments", tags=["assignments"])
|
|
|
|
|
|
async def _get_member_count(db: AsyncSession, class_id: int) -> int:
|
|
result = await db.execute(
|
|
select(func.count(User.id))
|
|
.join(Class_.memberships)
|
|
.where(Class_.id == class_id)
|
|
)
|
|
return result.scalar() or 0
|
|
|
|
|
|
def _build_assignment_out(a: any, user_id: int, total_members: int = 0) -> AssignmentOut:
|
|
submission_count = len(a.submissions) if a.submissions else 0
|
|
my_submitted = any(s.student_id == user_id for s in (a.submissions or []))
|
|
return AssignmentOut(
|
|
id=a.id,
|
|
class_id=a.class_id,
|
|
creator_id=a.creator_id,
|
|
creator_name=a.creator.name if a.creator else "Unknown",
|
|
title=a.title,
|
|
description=a.description,
|
|
deadline=a.deadline,
|
|
attachment_urls=a.get_attachment_urls_list(),
|
|
status=a.status,
|
|
submission_count=submission_count,
|
|
total_members=total_members,
|
|
my_submitted=my_submitted,
|
|
created_at=a.created_at,
|
|
updated_at=a.updated_at,
|
|
)
|
|
|
|
|
|
def _build_submission_out(s: any) -> SubmissionOut:
|
|
return SubmissionOut(
|
|
id=s.id,
|
|
assignment_id=s.assignment_id,
|
|
student_id=s.student_id,
|
|
student_name=s.student.name if s.student else "Unknown",
|
|
notes=s.notes,
|
|
file_url=s.file_url,
|
|
file_name=s.file_name,
|
|
file_type=s.file_type,
|
|
file_size=s.file_size,
|
|
grade=s.grade,
|
|
feedback=s.feedback,
|
|
graded_at=s.graded_at,
|
|
created_at=s.created_at,
|
|
updated_at=s.updated_at,
|
|
)
|
|
|
|
|
|
@router.get("/", response_model=PageResponse[AssignmentOut])
|
|
async def get_assignments(
|
|
page: int = 1,
|
|
page_size: int = 20,
|
|
class_id: int | None = None,
|
|
user: User = Depends(require_role("super_admin", "teacher", "student")),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
effective_class_id = resolve_class_id_for_user(user, class_id)
|
|
if effective_class_id is None:
|
|
return PageResponse(items=[], total=0, page=page, page_size=page_size, total_pages=0)
|
|
ensure_class_permission(user, "class_view", effective_class_id)
|
|
|
|
assignments, total = await list_assignments(db, effective_class_id, page, page_size)
|
|
total_pages = (total + page_size - 1) // page_size
|
|
member_count = await _get_member_count(db, effective_class_id)
|
|
items = [_build_assignment_out(a, user.id, member_count) for a in assignments]
|
|
return PageResponse(items=items, total=total, page=page, page_size=page_size, total_pages=total_pages)
|
|
|
|
|
|
@router.post("/", response_model=AssignmentOut)
|
|
async def create_new_assignment(
|
|
data: AssignmentCreate,
|
|
class_id: int | None = None,
|
|
user: User = Depends(require_role("super_admin", "teacher", "student")),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
effective_class_id = resolve_class_id_for_user(user, class_id)
|
|
if effective_class_id is None:
|
|
raise HTTPException(status_code=400, detail="You are not assigned to a class")
|
|
ensure_class_permission(user, "assignment_manage", effective_class_id)
|
|
|
|
assignment = await create_assignment(db, effective_class_id, user.id, data)
|
|
member_count = await _get_member_count(db, effective_class_id)
|
|
return _build_assignment_out(assignment, user.id, member_count)
|
|
|
|
|
|
@router.post("/{assignment_id}/attachments")
|
|
async def upload_assignment_attachments(
|
|
assignment_id: int,
|
|
files: list[UploadFile] = File(...),
|
|
user: User = Depends(require_role("super_admin", "teacher", "student")),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
assignment = await get_assignment_by_id(db, assignment_id)
|
|
if assignment is None:
|
|
raise HTTPException(status_code=404, detail="Assignment not found")
|
|
ensure_class_permission(user, "assignment_manage", assignment.class_id)
|
|
|
|
urls = []
|
|
for f in files:
|
|
contents = await f.read()
|
|
if len(contents) > 20 * 1024 * 1024: # 20MB limit
|
|
raise HTTPException(status_code=400, detail=f"File {f.filename} too large (max 20MB)")
|
|
url = upload_file(
|
|
f"assignments/{assignment_id}", f.filename or "file", contents,
|
|
f.content_type or "application/octet-stream",
|
|
)
|
|
urls.append(url)
|
|
|
|
await add_attachments(db, assignment, urls)
|
|
return {"attachment_urls": urls}
|
|
|
|
|
|
@router.get("/{assignment_id}", response_model=AssignmentDetailOut)
|
|
async def get_assignment_detail(
|
|
assignment_id: int,
|
|
user: User = Depends(require_role("super_admin", "teacher", "student")),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
assignment = await get_assignment_by_id(db, assignment_id)
|
|
if assignment is None:
|
|
raise HTTPException(status_code=404, detail="Assignment not found")
|
|
ensure_class_permission(user, "class_view", assignment.class_id)
|
|
|
|
base = _build_assignment_out(assignment, user.id, await _get_member_count(db, assignment.class_id))
|
|
|
|
# Student only sees their own submission
|
|
if "assignment_manage" not in get_effective_class_permissions(user, assignment.class_id) and user.role == "student":
|
|
my_submission = None
|
|
for s in (assignment.submissions or []):
|
|
if s.student_id == user.id:
|
|
my_submission = _build_submission_out(s)
|
|
break
|
|
return AssignmentDetailOut(**base.model_dump(), submissions=[my_submission] if my_submission else [])
|
|
|
|
# Admin sees all submissions
|
|
submissions = [_build_submission_out(s) for s in (assignment.submissions or [])]
|
|
return AssignmentDetailOut(**base.model_dump(), submissions=submissions)
|
|
|
|
|
|
@router.put("/{assignment_id}", response_model=AssignmentOut)
|
|
async def update_existing_assignment(
|
|
assignment_id: int,
|
|
data: AssignmentUpdate,
|
|
user: User = Depends(require_role("super_admin", "teacher", "student")),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
assignment = await get_assignment_by_id(db, assignment_id)
|
|
if assignment is None:
|
|
raise HTTPException(status_code=404, detail="Assignment not found")
|
|
ensure_class_permission(user, "assignment_manage", assignment.class_id)
|
|
|
|
updated = await update_assignment(db, assignment, data)
|
|
return _build_assignment_out(updated, user.id, await _get_member_count(db, updated.class_id))
|
|
|
|
|
|
@router.delete("/{assignment_id}")
|
|
async def delete_existing_assignment(
|
|
assignment_id: int,
|
|
user: User = Depends(require_role("super_admin", "teacher", "student")),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
assignment = await get_assignment_by_id(db, assignment_id)
|
|
if assignment is None:
|
|
raise HTTPException(status_code=404, detail="Assignment not found")
|
|
ensure_class_permission(user, "assignment_manage", assignment.class_id)
|
|
|
|
await delete_assignment(db, assignment)
|
|
return {"message": "Assignment deleted"}
|
|
|
|
|
|
@router.post("/{assignment_id}/submit", response_model=SubmissionOut)
|
|
async def submit_assignment(
|
|
assignment_id: int,
|
|
notes: str = "",
|
|
file: UploadFile = File(...),
|
|
user: User = Depends(require_role("super_admin", "teacher", "student")),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
assignment = await get_assignment_by_id(db, assignment_id)
|
|
if assignment is None:
|
|
raise HTTPException(status_code=404, detail="Assignment not found")
|
|
ensure_class_permission(user, "class_view", assignment.class_id)
|
|
|
|
# Upload file
|
|
file_url = None
|
|
file_name = None
|
|
file_type = None
|
|
file_size = None
|
|
if file.filename:
|
|
contents = await file.read()
|
|
if len(contents) > 50 * 1024 * 1024: # 50MB limit
|
|
raise HTTPException(status_code=400, detail="File too large (max 50MB)")
|
|
file_url = upload_file(
|
|
f"submissions/{assignment_id}/{user.id}", file.filename, contents,
|
|
file.content_type or "application/octet-stream",
|
|
)
|
|
file_name = file.filename
|
|
file_type = file.content_type
|
|
file_size = len(contents)
|
|
|
|
try:
|
|
submission = await create_submission(
|
|
db, assignment_id, user.id,
|
|
notes=notes or None,
|
|
file_url=file_url,
|
|
file_name=file_name,
|
|
file_type=file_type,
|
|
file_size=file_size,
|
|
)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
return _build_submission_out(submission)
|
|
|
|
|
|
@router.put("/submissions/{submission_id}/grade", response_model=SubmissionOut)
|
|
async def grade_assignment_submission(
|
|
submission_id: int,
|
|
data: SubmissionGrade,
|
|
user: User = Depends(require_role("super_admin", "teacher", "student")),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
from sqlalchemy import select
|
|
from app.db.models import AssignmentSubmission
|
|
result = await db.execute(
|
|
select(AssignmentSubmission)
|
|
.where(AssignmentSubmission.id == submission_id)
|
|
)
|
|
submission = result.scalar_one_or_none()
|
|
if submission is None:
|
|
raise HTTPException(status_code=404, detail="Submission not found")
|
|
assignment = await get_assignment_by_id(db, submission.assignment_id)
|
|
if assignment is None:
|
|
raise HTTPException(status_code=404, detail="Assignment not found")
|
|
ensure_class_permission(user, "assignment_manage", assignment.class_id)
|
|
|
|
graded = await grade_submission(db, submission, data)
|
|
|
|
# Reload with student relationship
|
|
from sqlalchemy.orm import selectinload
|
|
result = await db.execute(
|
|
select(AssignmentSubmission)
|
|
.options(selectinload(AssignmentSubmission.student))
|
|
.where(AssignmentSubmission.id == graded.id)
|
|
)
|
|
graded = result.scalar_one()
|
|
return _build_submission_out(graded)
|