70 lines
3.3 KiB
Python
70 lines
3.3 KiB
Python
from fastapi import HTTPException
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from job.assignment.models import ApplicationAssignments, JobAssignments
|
|
from job.assignment.serializers import serialize_application_assignment, serialize_job_assignment
|
|
from role.models import EnumRoles, Roles
|
|
from users.models import Users
|
|
|
|
|
|
class Assignment:
|
|
def __init__(self,session:AsyncSession):
|
|
self.session=session
|
|
|
|
async def _require_recruiter(self,user_id):
|
|
role=await Roles.get_role_by_name(self.session,EnumRoles.RECRUITER.value)
|
|
user=await Users.get_user_by_id(self.session,user_id)
|
|
if not role or not user or user.role_id!=role.id:
|
|
raise HTTPException(status_code=422,detail="user_id must be a recruiter")
|
|
return user
|
|
|
|
async def list_job_assignments(self,job_post_id):
|
|
if not job_post_id:
|
|
raise HTTPException(status_code=400,detail="job_post_id is required")
|
|
rows=await JobAssignments.fetch_by_job(self.session,job_post_id)
|
|
return [serialize_job_assignment(r) for r in rows]
|
|
|
|
async def create_job_assignment(self,payload,current_user):
|
|
user_id=payload.get("user_id")
|
|
job_post_id=payload.get("job_post_id")
|
|
if not user_id or not job_post_id:
|
|
raise HTTPException(status_code=422,detail="user_id and job_post_id are required")
|
|
await self._require_recruiter(user_id)
|
|
fields={
|
|
"job_post_id":JobAssignments._as_uuid(job_post_id),
|
|
"user_id":JobAssignments._as_uuid(user_id),
|
|
"assignment_role":payload.get("assignment_role") or "primary_recruiter",
|
|
"assigned_by":JobAssignments._as_uuid(
|
|
current_user.get("id") if isinstance(current_user,dict) else None
|
|
),
|
|
}
|
|
if not fields["job_post_id"] or not fields["user_id"] or not fields["assigned_by"]:
|
|
raise HTTPException(status_code=422,detail="Invalid job_post_id, user_id, or assigned_by")
|
|
row=await JobAssignments.insert_assignment(self.session,fields)
|
|
return serialize_job_assignment(row)
|
|
|
|
async def list_application_assignments(self,inbox_id):
|
|
if inbox_id is None:
|
|
raise HTTPException(status_code=400,detail="inbox_id is required")
|
|
rows=await ApplicationAssignments.fetch_by_inbox(self.session,int(inbox_id))
|
|
return [serialize_application_assignment(r) for r in rows]
|
|
|
|
async def create_application_assignment(self,payload,current_user):
|
|
user_id=payload.get("user_id")
|
|
inbox_id=payload.get("inbox_id")
|
|
if not user_id or inbox_id is None:
|
|
raise HTTPException(status_code=422,detail="user_id and inbox_id are required")
|
|
await self._require_recruiter(user_id)
|
|
fields={
|
|
"inbox_id":int(inbox_id),
|
|
"user_id":ApplicationAssignments._as_uuid(user_id),
|
|
"assignment_role":payload.get("assignment_role") or "primary_recruiter",
|
|
"assigned_by":ApplicationAssignments._as_uuid(
|
|
current_user.get("id") if isinstance(current_user,dict) else None
|
|
),
|
|
}
|
|
if not fields["user_id"] or not fields["assigned_by"]:
|
|
raise HTTPException(status_code=422,detail="Invalid user_id or assigned_by")
|
|
row=await ApplicationAssignments.insert_assignment(self.session,fields)
|
|
return serialize_application_assignment(row)
|