142 lines
6.6 KiB
Python
142 lines
6.6 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 job.job_post.models import JobPosts
|
|
from role.models import EnumRoles, Roles
|
|
from users.models import Users
|
|
|
|
# job_assignments.assignment_role → the users.role that may hold it.
|
|
# primary_recruiter is swappable; hiring_manager is the requisition owner.
|
|
JOB_ASSIGNMENT_ROLES = {
|
|
"primary_recruiter": EnumRoles.RECRUITER,
|
|
"hiring_manager": EnumRoles.HIRING_MANAGER,
|
|
}
|
|
JOB_OWNER_COLUMN = {
|
|
"primary_recruiter": "current_recruiter_id",
|
|
"hiring_manager": "hiring_manager_id",
|
|
}
|
|
|
|
|
|
class Assignment:
|
|
def __init__(self,session:AsyncSession):
|
|
self.session=session
|
|
|
|
async def require_role(self,user_id,role_enum,field_name):
|
|
role=await Roles.get_role_by_name(self.session,role_enum.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=f"{field_name} must be a {role_enum.value}",
|
|
)
|
|
if not user.is_active or user.is_deleted:
|
|
raise HTTPException(status_code=422,detail=f"{field_name} is not an active user")
|
|
return user
|
|
|
|
def _job_role(self,raw):
|
|
key=(raw or "primary_recruiter").strip()
|
|
if key=="recruiter":
|
|
key="primary_recruiter"
|
|
if key not in JOB_ASSIGNMENT_ROLES:
|
|
allowed=", ".join(sorted(JOB_ASSIGNMENT_ROLES))
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail=f"assignment_role must be one of {allowed}",
|
|
)
|
|
return key
|
|
|
|
async def record_job_owner(self,job_post_id,user_id,assignment_role,assigned_by):
|
|
"""Close the open interval of this role, then open a new one.
|
|
|
|
user_id None = unassign (hiring_manager cannot be cleared; callers
|
|
must not pass None for that role). No-ops when the same person already
|
|
holds the open interval. Does not touch job_posts columns.
|
|
"""
|
|
role=self._job_role(assignment_role)
|
|
job_uid=JobAssignments._as_uuid(job_post_id)
|
|
by_uid=JobAssignments._as_uuid(assigned_by)
|
|
if not job_uid or not by_uid:
|
|
raise HTTPException(status_code=422,detail="Invalid job_post_id or assigned_by")
|
|
current=await JobAssignments.fetch_by_job(
|
|
self.session,job_uid,current_only=True,assignment_role=role,
|
|
)
|
|
if user_id is None:
|
|
if role=="hiring_manager":
|
|
raise HTTPException(status_code=422,detail="hiring_manager_id is required")
|
|
await JobAssignments.close_current(self.session,job_uid,role)
|
|
return None
|
|
user_uid=JobAssignments._as_uuid(user_id)
|
|
if not user_uid:
|
|
raise HTTPException(status_code=422,detail="Invalid user_id")
|
|
if current and str(current[0].user_id)==str(user_uid):
|
|
return current[0]
|
|
await JobAssignments.close_current(self.session,job_uid,role)
|
|
return await JobAssignments.insert_assignment(self.session,{
|
|
"job_post_id":job_uid,
|
|
"user_id":user_uid,
|
|
"assignment_role":role,
|
|
"assigned_by":by_uid,
|
|
})
|
|
|
|
async def list_job_assignments(self,job_post_id,current_only=True,assignment_role=None):
|
|
if not job_post_id:
|
|
raise HTTPException(status_code=400,detail="job_post_id is required")
|
|
role=self._job_role(assignment_role) if assignment_role else None
|
|
rows=await JobAssignments.fetch_by_job(
|
|
self.session,job_post_id,current_only=current_only,assignment_role=role,
|
|
)
|
|
names=await Users.names_by_ids(
|
|
self.session,[r.user_id for r in rows]+[r.assigned_by for r in rows],
|
|
)
|
|
return [serialize_job_assignment(r,names=names) 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")
|
|
job=await JobPosts.get_job_post_by_id(self.session,job_post_id)
|
|
if not job or job.is_deleted:
|
|
raise HTTPException(status_code=404,detail="Job post not found")
|
|
role=self._job_role(payload.get("assignment_role"))
|
|
await self.require_role(user_id,JOB_ASSIGNMENT_ROLES[role],"user_id")
|
|
assigned_by=current_user.get("id") if isinstance(current_user,dict) else None
|
|
row=await self.record_job_owner(job_post_id,user_id,role,assigned_by)
|
|
column=JOB_OWNER_COLUMN[role]
|
|
await JobPosts.update_job_post(self.session,job_post_id,{column:user_id})
|
|
names=await Users.names_by_ids(
|
|
self.session,[row.user_id,row.assigned_by] if row else [],
|
|
)
|
|
return serialize_job_assignment(row,names=names) if row else None
|
|
|
|
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))
|
|
names=await Users.names_by_ids(
|
|
self.session,[r.user_id for r in rows]+[r.assigned_by for r in rows],
|
|
)
|
|
return [serialize_application_assignment(r,names=names) 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_role(user_id,EnumRoles.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)
|
|
names=await Users.names_by_ids(self.session,[row.user_id,row.assigned_by])
|
|
return serialize_application_assignment(row,names=names)
|