136 lines
4.6 KiB
Python
136 lines
4.6 KiB
Python
from datetime import datetime
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from fastapi.responses import JSONResponse
|
|
from pydantic import BaseModel
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from assessments.views import Assessment
|
|
from db_setup import get_session
|
|
from users.permissions import PermissionTag, require_permission
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
class AssessmentCreate(BaseModel):
|
|
assessment_type: str
|
|
inbox_id: int | None = None
|
|
manual_upload_candidate_id: str | None = None
|
|
job_post_id: str | None = None
|
|
duration_minutes: int | None = None
|
|
due_at: datetime | None = None
|
|
|
|
|
|
class AssessmentUpdate(BaseModel):
|
|
assessment_status: str | None = None
|
|
score: int | None = None
|
|
section_scores: list | None = None
|
|
due_at: datetime | None = None
|
|
|
|
|
|
@router.get("/assessments/fetch")
|
|
async def fetch_assessments(
|
|
current_user: dict = Depends(require_permission(PermissionTag.ASSESSMENTS_VIEW)),
|
|
assessment_id: str | None = Query(None),
|
|
inbox_id: int | None = Query(None),
|
|
manual_upload_candidate_id: str | None = Query(None),
|
|
job_post_id: str | None = Query(None),
|
|
assessment_status: str | None = Query(None),
|
|
top: int | None = Query(None),
|
|
skip: int = Query(0, ge=0),
|
|
session: AsyncSession = Depends(get_session),
|
|
):
|
|
try:
|
|
service = Assessment(session=session)
|
|
data, total = await service.get_assessments(
|
|
assessment_id, inbox_id, manual_upload_candidate_id, job_post_id,
|
|
assessment_status, top, skip,
|
|
)
|
|
return JSONResponse(content={"data": data, "total": total, "status_code": 200})
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.get("/assessments/counts")
|
|
async def fetch_assessment_counts(
|
|
current_user: dict = Depends(require_permission(PermissionTag.ASSESSMENTS_VIEW)),
|
|
session: AsyncSession = Depends(get_session),
|
|
):
|
|
try:
|
|
service = Assessment(session=session)
|
|
data = await service.get_counts()
|
|
return JSONResponse(content={"data": data, "status_code": 200})
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.post("/assessments/create")
|
|
async def create_assessment(
|
|
payload: AssessmentCreate,
|
|
current_user: dict = Depends(require_permission(PermissionTag.ASSESSMENTS_CREATE)),
|
|
session: AsyncSession = Depends(get_session),
|
|
):
|
|
try:
|
|
service = Assessment(session=session)
|
|
data = await service.create_assessment(payload.model_dump(exclude_unset=True), current_user)
|
|
return JSONResponse(content={"data": data, "status_code": 200})
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.patch("/assessments/update")
|
|
async def update_assessment(
|
|
payload: AssessmentUpdate,
|
|
current_user: dict = Depends(require_permission(PermissionTag.ASSESSMENTS_EDIT)),
|
|
assessment_id: str = Query(...),
|
|
session: AsyncSession = Depends(get_session),
|
|
):
|
|
try:
|
|
service = Assessment(session=session)
|
|
data = await service.update_assessment(
|
|
assessment_id, payload.model_dump(exclude_unset=True), current_user
|
|
)
|
|
return JSONResponse(content={"data": data, "status_code": 200})
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.delete("/assessments/delete")
|
|
async def delete_assessment(
|
|
current_user: dict = Depends(require_permission(PermissionTag.ASSESSMENTS_DELETE)),
|
|
assessment_id: str = Query(...),
|
|
session: AsyncSession = Depends(get_session),
|
|
):
|
|
try:
|
|
service = Assessment(session=session)
|
|
data = await service.delete_assessment(assessment_id, current_user)
|
|
return JSONResponse(content={"data": data, "status_code": 200})
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.post("/assessments/remind")
|
|
async def remind_assessment(
|
|
current_user: dict = Depends(require_permission(PermissionTag.ASSESSMENTS_EDIT)),
|
|
assessment_id: str = Query(...),
|
|
session: AsyncSession = Depends(get_session),
|
|
):
|
|
try:
|
|
service = Assessment(session=session)
|
|
data = await service.remind_assessment(assessment_id, current_user)
|
|
return JSONResponse(content={"data": data, "status_code": 200})
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|