mirror of
https://github.com/rzmk/learnhouse.git
synced 2025-12-19 04:19:25 +00:00
feat: add Assignments, Tasks, Submissions CRUD
This commit is contained in:
swve
parent
cd2397f4f7
commit
47782b57bc
32 changed files with 1719 additions and 218 deletions
997
apps/api/src/services/courses/activities/assignments.py
Normal file
997
apps/api/src/services/courses/activities/assignments.py
Normal file
|
|
@ -0,0 +1,997 @@
|
|||
####################################################
|
||||
# CRUD
|
||||
####################################################
|
||||
|
||||
from datetime import datetime
|
||||
from typing import Literal
|
||||
from uuid import uuid4
|
||||
from fastapi import HTTPException, Request
|
||||
from sqlmodel import Session, select
|
||||
|
||||
from src.db.courses.assignments import (
|
||||
Assignment,
|
||||
AssignmentCreate,
|
||||
AssignmentRead,
|
||||
AssignmentTask,
|
||||
AssignmentTaskCreate,
|
||||
AssignmentTaskRead,
|
||||
AssignmentTaskSubmission,
|
||||
AssignmentTaskSubmissionCreate,
|
||||
AssignmentTaskSubmissionRead,
|
||||
AssignmentTaskUpdate,
|
||||
AssignmentUpdate,
|
||||
AssignmentUserSubmission,
|
||||
AssignmentUserSubmissionCreate,
|
||||
AssignmentUserSubmissionRead,
|
||||
)
|
||||
from src.db.courses.courses import Course
|
||||
from src.db.users import AnonymousUser, PublicUser
|
||||
from src.security.rbac.rbac import (
|
||||
authorization_verify_based_on_roles_and_authorship_and_usergroups,
|
||||
authorization_verify_if_element_is_public,
|
||||
authorization_verify_if_user_is_anon,
|
||||
)
|
||||
|
||||
## > Assignments CRUD
|
||||
|
||||
|
||||
async def create_assignment(
|
||||
request: Request,
|
||||
assignment_object: AssignmentCreate,
|
||||
current_user: PublicUser | AnonymousUser,
|
||||
db_session: Session,
|
||||
):
|
||||
# Check if org exists
|
||||
statement = select(Course).where(Course.id == assignment_object.course_id)
|
||||
course = db_session.exec(statement).first()
|
||||
|
||||
if not course:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Course not found",
|
||||
)
|
||||
|
||||
# RBAC check
|
||||
await rbac_check(request, course.course_uuid, current_user, "create", db_session)
|
||||
|
||||
# Create Assignment
|
||||
assignment = Assignment(**assignment_object.model_dump())
|
||||
|
||||
assignment.assignment_uuid = str(f"assignment_{uuid4()}")
|
||||
assignment.creation_date = str(datetime.now())
|
||||
assignment.update_date = str(datetime.now())
|
||||
assignment.org_id = course.org_id
|
||||
|
||||
# Insert Assignment in DB
|
||||
db_session.add(assignment)
|
||||
db_session.commit()
|
||||
db_session.refresh(assignment)
|
||||
|
||||
# return assignment read
|
||||
return AssignmentRead.model_validate(assignment)
|
||||
|
||||
|
||||
async def read_assignment(
|
||||
request: Request,
|
||||
assignment_uuid: str,
|
||||
current_user: PublicUser | AnonymousUser,
|
||||
db_session: Session,
|
||||
):
|
||||
# Check if assignment exists
|
||||
statement = select(Assignment).where(Assignment.assignment_uuid == assignment_uuid)
|
||||
assignment = db_session.exec(statement).first()
|
||||
|
||||
if not assignment:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Assignment not found",
|
||||
)
|
||||
|
||||
# Check if course exists
|
||||
statement = select(Course).where(Course.id == assignment.course_id)
|
||||
course = db_session.exec(statement).first()
|
||||
|
||||
if not course:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Course not found",
|
||||
)
|
||||
|
||||
# RBAC check
|
||||
await rbac_check(request, course.course_uuid, current_user, "read", db_session)
|
||||
|
||||
# return assignment read
|
||||
return AssignmentRead.model_validate(assignment)
|
||||
|
||||
|
||||
async def update_assignment(
|
||||
request: Request,
|
||||
assignment_uuid: str,
|
||||
assignment_object: AssignmentUpdate,
|
||||
current_user: PublicUser | AnonymousUser,
|
||||
db_session: Session,
|
||||
):
|
||||
# Check if assignment exists
|
||||
statement = select(Assignment).where(Assignment.assignment_uuid == assignment_uuid)
|
||||
assignment = db_session.exec(statement).first()
|
||||
|
||||
if not assignment:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Assignment not found",
|
||||
)
|
||||
|
||||
# Check if course exists
|
||||
statement = select(Course).where(Course.id == assignment.course_id)
|
||||
course = db_session.exec(statement).first()
|
||||
|
||||
if not course:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Course not found",
|
||||
)
|
||||
|
||||
# RBAC check
|
||||
await rbac_check(request, course.course_uuid, current_user, "update", db_session)
|
||||
|
||||
# Update only the fields that were passed in
|
||||
for var, value in vars(assignment_object).items():
|
||||
if value is not None:
|
||||
setattr(assignment, var, value)
|
||||
assignment.update_date = str(datetime.now())
|
||||
|
||||
# Insert Assignment in DB
|
||||
db_session.add(assignment)
|
||||
db_session.commit()
|
||||
db_session.refresh(assignment)
|
||||
|
||||
# return assignment read
|
||||
return AssignmentRead.model_validate(assignment)
|
||||
|
||||
|
||||
async def delete_assignment(
|
||||
request: Request,
|
||||
assignment_uuid: str,
|
||||
current_user: PublicUser | AnonymousUser,
|
||||
db_session: Session,
|
||||
):
|
||||
# Check if assignment exists
|
||||
statement = select(Assignment).where(Assignment.assignment_uuid == assignment_uuid)
|
||||
assignment = db_session.exec(statement).first()
|
||||
|
||||
if not assignment:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Assignment not found",
|
||||
)
|
||||
|
||||
# Check if course exists
|
||||
statement = select(Course).where(Course.id == assignment.course_id)
|
||||
course = db_session.exec(statement).first()
|
||||
|
||||
if not course:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Course not found",
|
||||
)
|
||||
|
||||
# RBAC check
|
||||
await rbac_check(request, course.course_uuid, current_user, "delete", db_session)
|
||||
|
||||
# Delete Assignment
|
||||
db_session.delete(assignment)
|
||||
db_session.commit()
|
||||
|
||||
return {"message": "Assignment deleted"}
|
||||
|
||||
|
||||
## > Assignments Tasks CRUD
|
||||
|
||||
|
||||
async def create_assignment_task(
|
||||
request: Request,
|
||||
assignment_uuid: str,
|
||||
assignment_task_object: AssignmentTaskCreate,
|
||||
current_user: PublicUser | AnonymousUser,
|
||||
db_session: Session,
|
||||
):
|
||||
# Check if assignment exists
|
||||
statement = select(Assignment).where(Assignment.assignment_uuid == assignment_uuid)
|
||||
assignment = db_session.exec(statement).first()
|
||||
|
||||
if not assignment:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Assignment not found",
|
||||
)
|
||||
|
||||
# Check if course exists
|
||||
statement = select(Course).where(Course.id == assignment.course_id)
|
||||
course = db_session.exec(statement).first()
|
||||
|
||||
if not course:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Course not found",
|
||||
)
|
||||
|
||||
# RBAC check
|
||||
await rbac_check(request, course.course_uuid, current_user, "create", db_session)
|
||||
|
||||
# Create Assignment Task
|
||||
assignment_task = AssignmentTask(**assignment_task_object.model_dump())
|
||||
|
||||
assignment_task.assignment_task_uuid = str(f"assignmenttask_{uuid4()}")
|
||||
assignment_task.creation_date = str(datetime.now())
|
||||
assignment_task.update_date = str(datetime.now())
|
||||
assignment_task.org_id = course.org_id
|
||||
|
||||
# Insert Assignment Task in DB
|
||||
db_session.add(assignment_task)
|
||||
db_session.commit()
|
||||
db_session.refresh(assignment_task)
|
||||
|
||||
# return assignment task read
|
||||
return AssignmentTaskRead.model_validate(assignment_task)
|
||||
|
||||
|
||||
async def read_assignment_tasks(
|
||||
request: Request,
|
||||
assignment_uuid: str,
|
||||
current_user: PublicUser | AnonymousUser,
|
||||
db_session: Session,
|
||||
):
|
||||
# Find assignment
|
||||
statement = select(Assignment).where(Assignment.assignment_uuid == assignment_uuid)
|
||||
assignment = db_session.exec(statement).first()
|
||||
|
||||
if not assignment:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Assignment not found",
|
||||
)
|
||||
|
||||
# Check if course exists
|
||||
statement = select(Course).where(Course.id == assignment.course_id)
|
||||
course = db_session.exec(statement).first()
|
||||
|
||||
if not course:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Course not found",
|
||||
)
|
||||
|
||||
# Find assignments tasks for an assignment
|
||||
statement = select(AssignmentTask).where(
|
||||
assignment.assignment_uuid == assignment_uuid
|
||||
)
|
||||
|
||||
# RBAC check
|
||||
await rbac_check(request, course.course_uuid, current_user, "read", db_session)
|
||||
|
||||
# return assignment tasks read
|
||||
return [
|
||||
AssignmentTaskRead.model_validate(assignment_task)
|
||||
for assignment_task in db_session.exec(statement).all()
|
||||
]
|
||||
|
||||
|
||||
async def update_assignment_task(
|
||||
request: Request,
|
||||
assignment_task_uuid: str,
|
||||
assignment_task_object: AssignmentTaskUpdate,
|
||||
current_user: PublicUser | AnonymousUser,
|
||||
db_session: Session,
|
||||
):
|
||||
# Check if assignment task exists
|
||||
statement = select(AssignmentTask).where(
|
||||
AssignmentTask.assignment_task_uuid == assignment_task_uuid
|
||||
)
|
||||
assignment_task = db_session.exec(statement).first()
|
||||
|
||||
if not assignment_task:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Assignment Task not found",
|
||||
)
|
||||
|
||||
# Check if assignment exists
|
||||
statement = select(Assignment).where(Assignment.id == assignment_task.assignment_id)
|
||||
assignment = db_session.exec(statement).first()
|
||||
|
||||
if not assignment:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Assignment not found",
|
||||
)
|
||||
|
||||
# Check if course exists
|
||||
statement = select(Course).where(Course.id == assignment.course_id)
|
||||
course = db_session.exec(statement).first()
|
||||
|
||||
if not course:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Course not found",
|
||||
)
|
||||
|
||||
# RBAC check
|
||||
await rbac_check(request, course.course_uuid, current_user, "update", db_session)
|
||||
|
||||
# Update only the fields that were passed in
|
||||
for var, value in vars(assignment_task_object).items():
|
||||
if value is not None:
|
||||
setattr(assignment_task, var, value)
|
||||
assignment_task.update_date = str(datetime.now())
|
||||
|
||||
# Insert Assignment Task in DB
|
||||
db_session.add(assignment_task)
|
||||
db_session.commit()
|
||||
db_session.refresh(assignment_task)
|
||||
|
||||
# return assignment task read
|
||||
return AssignmentTaskRead.model_validate(assignment_task)
|
||||
|
||||
|
||||
async def delete_assignment_task(
|
||||
request: Request,
|
||||
assignment_task_uuid: str,
|
||||
current_user: PublicUser | AnonymousUser,
|
||||
db_session: Session,
|
||||
):
|
||||
# Check if assignment task exists
|
||||
statement = select(AssignmentTask).where(
|
||||
AssignmentTask.assignment_task_uuid == assignment_task_uuid
|
||||
)
|
||||
assignment_task = db_session.exec(statement).first()
|
||||
|
||||
if not assignment_task:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Assignment Task not found",
|
||||
)
|
||||
|
||||
# Check if assignment exists
|
||||
statement = select(Assignment).where(Assignment.id == assignment_task.assignment_id)
|
||||
assignment = db_session.exec(statement).first()
|
||||
|
||||
if not assignment:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Assignment not found",
|
||||
)
|
||||
|
||||
# Check if course exists
|
||||
statement = select(Course).where(Course.id == assignment.course_id)
|
||||
course = db_session.exec(statement).first()
|
||||
|
||||
if not course:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Course not found",
|
||||
)
|
||||
|
||||
# RBAC check
|
||||
await rbac_check(request, course.course_uuid, current_user, "delete", db_session)
|
||||
|
||||
# Delete Assignment Task
|
||||
db_session.delete(assignment_task)
|
||||
db_session.commit()
|
||||
|
||||
return {"message": "Assignment Task deleted"}
|
||||
|
||||
|
||||
## > Assignments Tasks Submissions CRUD
|
||||
|
||||
|
||||
async def create_assignment_task_submission(
|
||||
request: Request,
|
||||
assignment_task_uuid: str,
|
||||
assignment_task_submission_object: AssignmentTaskSubmissionCreate,
|
||||
current_user: PublicUser | AnonymousUser,
|
||||
db_session: Session,
|
||||
):
|
||||
# Check if assignment task exists
|
||||
statement = select(AssignmentTask).where(
|
||||
AssignmentTask.assignment_task_uuid == assignment_task_uuid
|
||||
)
|
||||
assignment_task = db_session.exec(statement).first()
|
||||
|
||||
if not assignment_task:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Assignment Task not found",
|
||||
)
|
||||
|
||||
# Check if assignment exists
|
||||
statement = select(Assignment).where(Assignment.id == assignment_task.assignment_id)
|
||||
assignment = db_session.exec(statement).first()
|
||||
|
||||
if not assignment:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Assignment not found",
|
||||
)
|
||||
|
||||
# Check if course exists
|
||||
statement = select(Course).where(Course.id == assignment.course_id)
|
||||
course = db_session.exec(statement).first()
|
||||
|
||||
if not course:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Course not found",
|
||||
)
|
||||
|
||||
# RBAC check
|
||||
await rbac_check(request, course.course_uuid, current_user, "create", db_session)
|
||||
|
||||
# Create Assignment Task Submission
|
||||
assignment_task_submission = AssignmentTaskSubmission(
|
||||
**assignment_task_submission_object.model_dump()
|
||||
)
|
||||
|
||||
assignment_task_submission.assignment_task_submission_uuid = str(
|
||||
f"assignmenttasksubmission_{uuid4()}"
|
||||
)
|
||||
assignment_task_submission.creation_date = str(datetime.now())
|
||||
assignment_task_submission.update_date = str(datetime.now())
|
||||
assignment_task_submission.org_id = course.org_id
|
||||
|
||||
# Insert Assignment Task Submission in DB
|
||||
db_session.add(assignment_task_submission)
|
||||
db_session.commit()
|
||||
db_session.refresh(assignment_task_submission)
|
||||
|
||||
# return assignment task submission read
|
||||
return AssignmentTaskSubmissionRead.model_validate(assignment_task_submission)
|
||||
|
||||
|
||||
async def read_user_assignment_task_submissions(
|
||||
request: Request,
|
||||
assignment_task_submission_uuid: str,
|
||||
user_id: int,
|
||||
current_user: PublicUser | AnonymousUser,
|
||||
db_session: Session,
|
||||
):
|
||||
# Check if assignment task submission exists
|
||||
statement = select(AssignmentTaskSubmission).where(
|
||||
AssignmentTaskSubmission.assignment_task_submission_uuid
|
||||
== assignment_task_submission_uuid,
|
||||
AssignmentTaskSubmission.user_id == user_id,
|
||||
)
|
||||
assignment_task_submission = db_session.exec(statement).first()
|
||||
|
||||
if not assignment_task_submission:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Assignment Task Submission not found",
|
||||
)
|
||||
|
||||
# Check if assignment task exists
|
||||
statement = select(AssignmentTask).where(
|
||||
AssignmentTask.id == assignment_task_submission.assignment_task_id
|
||||
)
|
||||
assignment_task = db_session.exec(statement).first()
|
||||
|
||||
if not assignment_task:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Assignment Task not found",
|
||||
)
|
||||
|
||||
# Check if assignment exists
|
||||
statement = select(Assignment).where(Assignment.id == assignment_task.assignment_id)
|
||||
assignment = db_session.exec(statement).first()
|
||||
|
||||
if not assignment:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Assignment not found",
|
||||
)
|
||||
|
||||
# Check if course exists
|
||||
statement = select(Course).where(Course.id == assignment.course_id)
|
||||
course = db_session.exec(statement).first()
|
||||
|
||||
if not course:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Course not found",
|
||||
)
|
||||
|
||||
# RBAC check
|
||||
await rbac_check(request, course.course_uuid, current_user, "read", db_session)
|
||||
|
||||
# return assignment task submission read
|
||||
return AssignmentTaskSubmissionRead.model_validate(assignment_task_submission)
|
||||
|
||||
|
||||
async def read_assignment_task_submissions(
|
||||
request: Request,
|
||||
assignment_task_submission_uuid: str,
|
||||
current_user: PublicUser | AnonymousUser,
|
||||
db_session: Session,
|
||||
):
|
||||
# Check if assignment task submission exists
|
||||
statement = select(AssignmentTaskSubmission).where(
|
||||
AssignmentTaskSubmission.assignment_task_submission_uuid
|
||||
== assignment_task_submission_uuid,
|
||||
)
|
||||
assignment_task_submission = db_session.exec(statement).first()
|
||||
|
||||
if not assignment_task_submission:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Assignment Task Submission not found",
|
||||
)
|
||||
|
||||
# Check if assignment task exists
|
||||
statement = select(AssignmentTask).where(
|
||||
AssignmentTask.id == assignment_task_submission.assignment_task_id
|
||||
)
|
||||
assignment_task = db_session.exec(statement).first()
|
||||
|
||||
if not assignment_task:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Assignment Task not found",
|
||||
)
|
||||
|
||||
# Check if assignment exists
|
||||
statement = select(Assignment).where(Assignment.id == assignment_task.assignment_id)
|
||||
assignment = db_session.exec(statement).first()
|
||||
|
||||
if not assignment:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Assignment not found",
|
||||
)
|
||||
|
||||
# Check if course exists
|
||||
statement = select(Course).where(Course.id == assignment.course_id)
|
||||
course = db_session.exec(statement).first()
|
||||
|
||||
if not course:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Course not found",
|
||||
)
|
||||
|
||||
# RBAC check
|
||||
await rbac_check(request, course.course_uuid, current_user, "read", db_session)
|
||||
|
||||
# return assignment task submission read
|
||||
return AssignmentTaskSubmissionRead.model_validate(assignment_task_submission)
|
||||
|
||||
|
||||
async def update_assignment_task_submission(
|
||||
request: Request,
|
||||
assignment_task_submission_uuid: str,
|
||||
assignment_task_submission_object: AssignmentTaskSubmissionCreate,
|
||||
current_user: PublicUser | AnonymousUser,
|
||||
db_session: Session,
|
||||
):
|
||||
# Check if assignment task submission exists
|
||||
statement = select(AssignmentTaskSubmission).where(
|
||||
AssignmentTaskSubmission.assignment_task_submission_uuid
|
||||
== assignment_task_submission_uuid
|
||||
)
|
||||
assignment_task_submission = db_session.exec(statement).first()
|
||||
|
||||
if not assignment_task_submission:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Assignment Task Submission not found",
|
||||
)
|
||||
|
||||
# Check if assignment task exists
|
||||
statement = select(AssignmentTask).where(
|
||||
AssignmentTask.id == assignment_task_submission.assignment_task_id
|
||||
)
|
||||
assignment_task = db_session.exec(statement).first()
|
||||
|
||||
if not assignment_task:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Assignment Task not found",
|
||||
)
|
||||
|
||||
# Check if assignment exists
|
||||
statement = select(Assignment).where(Assignment.id == assignment_task.assignment_id)
|
||||
assignment = db_session.exec(statement).first()
|
||||
|
||||
if not assignment:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Assignment not found",
|
||||
)
|
||||
|
||||
# Check if course exists
|
||||
statement = select(Course).where(Course.id == assignment.course_id)
|
||||
course = db_session.exec(statement).first()
|
||||
|
||||
if not course:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Course not found",
|
||||
)
|
||||
|
||||
# RBAC check
|
||||
await rbac_check(request, course.course_uuid, current_user, "update", db_session)
|
||||
|
||||
# Update only the fields that were passed in
|
||||
for var, value in vars(assignment_task_submission_object).items():
|
||||
if value is not None:
|
||||
setattr(assignment_task_submission, var, value)
|
||||
assignment_task_submission.update_date = str(datetime.now())
|
||||
|
||||
# Insert Assignment Task Submission in DB
|
||||
db_session.add(assignment_task_submission)
|
||||
db_session.commit()
|
||||
db_session.refresh(assignment_task_submission)
|
||||
|
||||
# return assignment task submission read
|
||||
return AssignmentTaskSubmissionRead.model_validate(assignment_task_submission)
|
||||
|
||||
|
||||
async def delete_assignment_task_submission(
|
||||
request: Request,
|
||||
assignment_task_submission_uuid: str,
|
||||
current_user: PublicUser | AnonymousUser,
|
||||
db_session: Session,
|
||||
):
|
||||
# Check if assignment task submission exists
|
||||
statement = select(AssignmentTaskSubmission).where(
|
||||
AssignmentTaskSubmission.assignment_task_submission_uuid
|
||||
== assignment_task_submission_uuid
|
||||
)
|
||||
assignment_task_submission = db_session.exec(statement).first()
|
||||
|
||||
if not assignment_task_submission:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Assignment Task Submission not found",
|
||||
)
|
||||
|
||||
# Check if assignment task exists
|
||||
statement = select(AssignmentTask).where(
|
||||
AssignmentTask.id == assignment_task_submission.assignment_task_id
|
||||
)
|
||||
assignment_task = db_session.exec(statement).first()
|
||||
|
||||
if not assignment_task:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Assignment Task not found",
|
||||
)
|
||||
|
||||
# Check if assignment exists
|
||||
statement = select(Assignment).where(Assignment.id == assignment_task.assignment_id)
|
||||
assignment = db_session.exec(statement).first()
|
||||
|
||||
if not assignment:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Assignment not found",
|
||||
)
|
||||
|
||||
# Check if course exists
|
||||
statement = select(Course).where(Course.id == assignment.course_id)
|
||||
course = db_session.exec(statement).first()
|
||||
|
||||
if not course:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Course not found",
|
||||
)
|
||||
|
||||
# RBAC check
|
||||
await rbac_check(request, course.course_uuid, current_user, "delete", db_session)
|
||||
|
||||
# Delete Assignment Task Submission
|
||||
db_session.delete(assignment_task_submission)
|
||||
db_session.commit()
|
||||
|
||||
return {"message": "Assignment Task Submission deleted"}
|
||||
|
||||
|
||||
## > Assignments Submissions CRUD
|
||||
|
||||
|
||||
async def create_assignment_submission(
|
||||
request: Request,
|
||||
assignment_uuid: str,
|
||||
assignment_user_submission_object: AssignmentUserSubmissionCreate,
|
||||
current_user: PublicUser | AnonymousUser,
|
||||
db_session: Session,
|
||||
):
|
||||
# Check if assignment exists
|
||||
statement = select(Assignment).where(Assignment.assignment_uuid == assignment_uuid)
|
||||
assignment = db_session.exec(statement).first()
|
||||
|
||||
if not assignment:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Assignment not found",
|
||||
)
|
||||
|
||||
# Check if the submission has already been made
|
||||
statement = select(AssignmentUserSubmission).where(
|
||||
AssignmentUserSubmission.assignment_id == assignment.id,
|
||||
AssignmentUserSubmission.user_id == assignment_user_submission_object.user_id,
|
||||
)
|
||||
|
||||
assignment_user_submission = db_session.exec(statement).first()
|
||||
|
||||
if assignment_user_submission:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail="Assignment User Submission already exists",
|
||||
)
|
||||
|
||||
# Check if course exists
|
||||
statement = select(Course).where(Course.id == assignment.course_id)
|
||||
course = db_session.exec(statement).first()
|
||||
|
||||
if not course:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Course not found",
|
||||
)
|
||||
|
||||
# RBAC check
|
||||
await rbac_check(request, course.course_uuid, current_user, "create", db_session)
|
||||
|
||||
# Create Assignment User Submission
|
||||
assignment_user_submission = AssignmentUserSubmission(
|
||||
**assignment_user_submission_object.model_dump()
|
||||
)
|
||||
|
||||
assignment_user_submission.assignment_user_submission_uuid = str(
|
||||
f"assignmentusersubmission_{uuid4()}"
|
||||
)
|
||||
assignment_user_submission.creation_date = str(datetime.now())
|
||||
assignment_user_submission.update_date = str(datetime.now())
|
||||
assignment_user_submission.org_id = course.org_id
|
||||
|
||||
# Insert Assignment User Submission in DB
|
||||
db_session.add(assignment_user_submission)
|
||||
db_session.commit()
|
||||
|
||||
# return assignment user submission read
|
||||
return AssignmentUserSubmissionRead.model_validate(assignment_user_submission)
|
||||
|
||||
|
||||
async def read_assignment_submissions(
|
||||
request: Request,
|
||||
assignment_uuid: str,
|
||||
current_user: PublicUser | AnonymousUser,
|
||||
db_session: Session,
|
||||
):
|
||||
# Find assignment
|
||||
statement = select(Assignment).where(Assignment.assignment_uuid == assignment_uuid)
|
||||
assignment = db_session.exec(statement).first()
|
||||
|
||||
if not assignment:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Assignment not found",
|
||||
)
|
||||
|
||||
# Check if course exists
|
||||
statement = select(Course).where(Course.id == assignment.course_id)
|
||||
course = db_session.exec(statement).first()
|
||||
|
||||
if not course:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Course not found",
|
||||
)
|
||||
|
||||
# Find assignments tasks for an assignment
|
||||
statement = select(AssignmentUserSubmission).where(
|
||||
assignment.assignment_uuid == assignment_uuid
|
||||
)
|
||||
|
||||
# RBAC check
|
||||
await rbac_check(request, course.course_uuid, current_user, "read", db_session)
|
||||
|
||||
# return assignment tasks read
|
||||
return [
|
||||
AssignmentUserSubmissionRead.model_validate(assignment_user_submission)
|
||||
for assignment_user_submission in db_session.exec(statement).all()
|
||||
]
|
||||
|
||||
|
||||
async def read_user_assignment_submissions(
|
||||
request: Request,
|
||||
assignment_uuid: str,
|
||||
user_id: int,
|
||||
current_user: PublicUser | AnonymousUser,
|
||||
db_session: Session,
|
||||
):
|
||||
# Find assignment
|
||||
statement = select(Assignment).where(Assignment.assignment_uuid == assignment_uuid)
|
||||
assignment = db_session.exec(statement).first()
|
||||
|
||||
if not assignment:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Assignment not found",
|
||||
)
|
||||
|
||||
# Check if course exists
|
||||
statement = select(Course).where(Course.id == assignment.course_id)
|
||||
course = db_session.exec(statement).first()
|
||||
|
||||
if not course:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Course not found",
|
||||
)
|
||||
|
||||
# Find assignments tasks for an assignment
|
||||
statement = select(AssignmentUserSubmission).where(
|
||||
assignment.assignment_uuid == assignment_uuid,
|
||||
AssignmentUserSubmission.user_id == user_id,
|
||||
)
|
||||
|
||||
# RBAC check
|
||||
await rbac_check(request, course.course_uuid, current_user, "read", db_session)
|
||||
|
||||
# return assignment tasks read
|
||||
return [
|
||||
AssignmentUserSubmissionRead.model_validate(assignment_user_submission)
|
||||
for assignment_user_submission in db_session.exec(statement).all()
|
||||
]
|
||||
|
||||
|
||||
async def update_assignment_submission(
|
||||
request: Request,
|
||||
user_id: str,
|
||||
assignment_user_submission_object: AssignmentUserSubmissionCreate,
|
||||
current_user: PublicUser | AnonymousUser,
|
||||
db_session: Session,
|
||||
):
|
||||
# Check if assignment user submission exists
|
||||
statement = select(AssignmentUserSubmission).where(
|
||||
AssignmentUserSubmission.user_id == user_id
|
||||
)
|
||||
assignment_user_submission = db_session.exec(statement).first()
|
||||
|
||||
if not assignment_user_submission:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Assignment User Submission not found",
|
||||
)
|
||||
|
||||
# Check if assignment exists
|
||||
statement = select(Assignment).where(
|
||||
Assignment.id == assignment_user_submission.assignment_id
|
||||
)
|
||||
assignment = db_session.exec(statement).first()
|
||||
|
||||
if not assignment:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Assignment not found",
|
||||
)
|
||||
|
||||
# Check if course exists
|
||||
statement = select(Course).where(Course.id == assignment.course_id)
|
||||
course = db_session.exec(statement).first()
|
||||
|
||||
if not course:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Course not found",
|
||||
)
|
||||
|
||||
# RBAC check
|
||||
await rbac_check(request, course.course_uuid, current_user, "update", db_session)
|
||||
|
||||
# Update only the fields that were passed in
|
||||
for var, value in vars(assignment_user_submission_object).items():
|
||||
if value is not None:
|
||||
setattr(assignment_user_submission, var, value)
|
||||
assignment_user_submission.update_date = str(datetime.now())
|
||||
|
||||
# Insert Assignment User Submission in DB
|
||||
db_session.add(assignment_user_submission)
|
||||
db_session.commit()
|
||||
db_session.refresh(assignment_user_submission)
|
||||
|
||||
# return assignment user submission read
|
||||
return AssignmentUserSubmissionRead.model_validate(assignment_user_submission)
|
||||
|
||||
|
||||
async def delete_assignment_submission(
|
||||
request: Request,
|
||||
user_id: str,
|
||||
assignment_id: str,
|
||||
current_user: PublicUser | AnonymousUser,
|
||||
db_session: Session,
|
||||
):
|
||||
# Check if assignment user submission exists
|
||||
statement = select(AssignmentUserSubmission).where(
|
||||
AssignmentUserSubmission.user_id == user_id,
|
||||
AssignmentUserSubmission.assignment_id == assignment_id,
|
||||
)
|
||||
assignment_user_submission = db_session.exec(statement).first()
|
||||
|
||||
if not assignment_user_submission:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Assignment User Submission not found",
|
||||
)
|
||||
|
||||
# Check if assignment exists
|
||||
statement = select(Assignment).where(
|
||||
Assignment.id == assignment_user_submission.assignment_id
|
||||
)
|
||||
assignment = db_session.exec(statement).first()
|
||||
|
||||
if not assignment:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Assignment not found",
|
||||
)
|
||||
|
||||
# Check if course exists
|
||||
statement = select(Course).where(Course.id == assignment.course_id)
|
||||
course = db_session.exec(statement).first()
|
||||
|
||||
if not course:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Course not found",
|
||||
)
|
||||
|
||||
# RBAC check
|
||||
await rbac_check(request, course.course_uuid, current_user, "delete", db_session)
|
||||
|
||||
# Delete Assignment User Submission
|
||||
db_session.delete(assignment_user_submission)
|
||||
db_session.commit()
|
||||
|
||||
return {"message": "Assignment User Submission deleted"}
|
||||
|
||||
|
||||
## 🔒 RBAC Utils ##
|
||||
|
||||
|
||||
async def rbac_check(
|
||||
request: Request,
|
||||
course_uuid: str,
|
||||
current_user: PublicUser | AnonymousUser,
|
||||
action: Literal["create", "read", "update", "delete"],
|
||||
db_session: Session,
|
||||
):
|
||||
|
||||
if action == "read":
|
||||
if current_user.id == 0: # Anonymous user
|
||||
res = await authorization_verify_if_element_is_public(
|
||||
request, course_uuid, action, db_session
|
||||
)
|
||||
return res
|
||||
else:
|
||||
res = (
|
||||
await authorization_verify_based_on_roles_and_authorship_and_usergroups(
|
||||
request, current_user.id, action, course_uuid, db_session
|
||||
)
|
||||
)
|
||||
return res
|
||||
else:
|
||||
await authorization_verify_if_user_is_anon(current_user.id)
|
||||
|
||||
await authorization_verify_based_on_roles_and_authorship_and_usergroups(
|
||||
request,
|
||||
current_user.id,
|
||||
action,
|
||||
course_uuid,
|
||||
db_session,
|
||||
)
|
||||
|
||||
|
||||
## 🔒 RBAC Utils ##
|
||||
Loading…
Add table
Add a link
Reference in a new issue