72 lines
3.2 KiB
Python
72 lines
3.2 KiB
Python
from fastapi import HTTPException
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from job.candidate.models import Interviews
|
|
from job.interviews.serializers import serialize_interview
|
|
from job.job_post.models import JobPosts
|
|
|
|
|
|
class Interview:
|
|
def __init__(self,session:AsyncSession):
|
|
self.session=session
|
|
|
|
async def _job_title_for(self,row):
|
|
inbox=getattr(row,"inbox",None)
|
|
messages=getattr(inbox,"messages",None) if inbox else None
|
|
job_post_id=getattr(messages,"assigned_job_post_id",None) if messages else None
|
|
if not job_post_id:
|
|
return None
|
|
job=await JobPosts.get_job_post_by_id(self.session,str(job_post_id))
|
|
return job.title if job else None
|
|
|
|
async def _serialize(self,row):
|
|
return serialize_interview(row,job_title=await self._job_title_for(row))
|
|
|
|
async def get_interview(self,interview_id=None,inbox_id=None,from_date=None,to_date=None,status=None,top=None,skip=0):
|
|
if interview_id:
|
|
row=await Interviews.get_interview_by_id(self.session,interview_id)
|
|
if not row:
|
|
raise HTTPException(status_code=404,detail="Interview not found")
|
|
return serialize_interview(row)
|
|
if inbox_id is not None:
|
|
rows=await Interviews.get_interviews_by_inbox(self.session,int(inbox_id))
|
|
return [serialize_interview(r) for r in rows]
|
|
if from_date is not None or to_date is not None or status is not None or top is not None:
|
|
return await self.get_interviews_range(
|
|
from_date=from_date,to_date=to_date,status=status,top=top,skip=skip,
|
|
)
|
|
raise HTTPException(status_code=400,detail="interview_id or inbox_id is required")
|
|
|
|
async def get_interviews_range(self,from_date=None,to_date=None,status=None,top=None,skip=0):
|
|
rows,total=await Interviews.get_interviews_in_range(
|
|
self.session,
|
|
from_date=from_date,
|
|
to_date=to_date,
|
|
status=status,
|
|
top=top,
|
|
skip=skip,
|
|
)
|
|
titles=await Interviews.job_titles_by_inbox(self.session,[r.inbox_id for r in rows])
|
|
return [serialize_interview(r,job_title=titles.get(r.inbox_id)) for r in rows],total
|
|
|
|
async def create_interview(self,payload):
|
|
fields={
|
|
"interview_date":payload.get("interview_date"),
|
|
"interview_time":payload.get("interview_time"),
|
|
"interview_type":payload.get("interview_type") or "",
|
|
"interview_status":payload.get("interview_status") or "",
|
|
"inbox_id":payload.get("inbox_id"),
|
|
}
|
|
fields={k:v for k,v in fields.items() if v is not None}
|
|
row=await Interviews.insert_interview(self.session,fields)
|
|
return await self._serialize(row)
|
|
|
|
async def update_interview(self,interview_id,payload):
|
|
fields={k:v for k,v in payload.items() if v is not None}
|
|
if not fields:
|
|
raise HTTPException(status_code=400,detail="No fields to update")
|
|
row=await Interviews.update_interview(self.session,interview_id,fields)
|
|
if not row:
|
|
raise HTTPException(status_code=404,detail="Interview not found")
|
|
return await self._serialize(row)
|