mirror of
https://github.com/rzmk/learnhouse.git
synced 2025-12-19 04:19:25 +00:00
1215 lines
36 KiB
Python
1215 lines
36 KiB
Python
####################################################
|
|
# CRUD
|
|
####################################################
|
|
|
|
from datetime import datetime
|
|
from typing import Literal
|
|
from uuid import uuid4
|
|
from fastapi import HTTPException, Request, UploadFile
|
|
from sqlmodel import Session, select
|
|
|
|
from src.db.courses.activities import Activity
|
|
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.organizations import Organization
|
|
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,
|
|
)
|
|
from src.services.courses.activities.uploads.tasks_ref_files import (
|
|
upload_reference_file,
|
|
)
|
|
|
|
## > 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 read_assignment_from_activity_uuid(
|
|
request: Request,
|
|
activity_uuid: str,
|
|
current_user: PublicUser | AnonymousUser,
|
|
db_session: Session,
|
|
):
|
|
# Check if activity exists
|
|
statement = select(Activity).where(Activity.activity_uuid == activity_uuid)
|
|
activity = db_session.exec(statement).first()
|
|
|
|
if not activity:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="Activity not found",
|
|
)
|
|
|
|
# Check if course exists
|
|
statement = select(Course).where(Course.id == activity.course_id)
|
|
course = db_session.exec(statement).first()
|
|
|
|
if not course:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="Course not found",
|
|
)
|
|
|
|
# Check if assignment exists
|
|
statement = select(Assignment).where(Assignment.activity_id == activity.id)
|
|
assignment = db_session.exec(statement).first()
|
|
|
|
if not assignment:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="Assignment 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"}
|
|
|
|
|
|
async def delete_assignment_from_activity_uuid(
|
|
request: Request,
|
|
activity_uuid: str,
|
|
current_user: PublicUser | AnonymousUser,
|
|
db_session: Session,
|
|
):
|
|
# Check if activity exists
|
|
statement = select(Activity).where(Activity.activity_uuid == activity_uuid)
|
|
|
|
activity = db_session.exec(statement).first()
|
|
|
|
if not activity:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="Activity not found",
|
|
)
|
|
|
|
# Check if course exists
|
|
statement = select(Course).where(Course.id == activity.course_id)
|
|
course = db_session.exec(statement).first()
|
|
|
|
if not course:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="Course not found",
|
|
)
|
|
|
|
# Check if assignment exists
|
|
statement = select(Assignment).where(Assignment.activity_id == activity.id)
|
|
assignment = db_session.exec(statement).first()
|
|
|
|
if not assignment:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="Assignment 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
|
|
assignment_task.chapter_id = assignment.chapter_id
|
|
assignment_task.activity_id = assignment.activity_id
|
|
assignment_task.assignment_id = assignment.id # type: ignore
|
|
assignment_task.course_id = assignment.course_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 read_assignment_task(
|
|
request: Request,
|
|
assignment_task_uuid: str,
|
|
current_user: PublicUser | AnonymousUser,
|
|
db_session: Session,
|
|
):
|
|
# Find assignment
|
|
statement = select(AssignmentTask).where(
|
|
AssignmentTask.assignment_task_uuid == assignment_task_uuid
|
|
)
|
|
assignmenttask = db_session.exec(statement).first()
|
|
|
|
if not assignmenttask:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="Assignment Task not found",
|
|
)
|
|
|
|
# Check if assignment exists
|
|
statement = select(Assignment).where(Assignment.id == assignmenttask.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 read
|
|
return AssignmentTaskRead.model_validate(assignmenttask)
|
|
|
|
|
|
async def put_assignment_task_reference_file(
|
|
request: Request,
|
|
db_session: Session,
|
|
assignment_task_uuid: str,
|
|
current_user: PublicUser | AnonymousUser,
|
|
reference_file: UploadFile | None = None,
|
|
):
|
|
# 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 for activity
|
|
statement = select(Activity).where(Activity.id == assignment.activity_id)
|
|
activity = db_session.exec(statement).first()
|
|
|
|
# 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",
|
|
)
|
|
|
|
# Get org uuid
|
|
org_statement = select(Organization).where(Organization.id == course.org_id)
|
|
org = db_session.exec(org_statement).first()
|
|
|
|
# RBAC check
|
|
await rbac_check(request, course.course_uuid, current_user, "update", db_session)
|
|
|
|
# Upload reference file
|
|
if reference_file and reference_file.filename and activity and org:
|
|
name_in_disk = (
|
|
f"{assignment_task_uuid}{uuid4()}.{reference_file.filename.split('.')[-1]}"
|
|
)
|
|
await upload_reference_file(
|
|
reference_file, name_in_disk, activity.activity_uuid, org.org_uuid, course.course_uuid, assignment.assignment_uuid, assignment_task_uuid
|
|
)
|
|
course.thumbnail_image = name_in_disk
|
|
# Update reference file
|
|
assignment_task.reference_file = name_in_disk
|
|
|
|
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 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 ##
|