61 lines
2.9 KiB
Python
61 lines
2.9 KiB
Python
from fastapi import HTTPException
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from candidate_forms.plugins import _as_uuid, _aware, _user_id
|
|
from reference_check.models import ReferenceCheck
|
|
from reference_check.serializers import serialize_reference_check
|
|
from users.models import Users
|
|
from users.permissions import is_admin
|
|
|
|
|
|
class ReferenceCheckForm:
|
|
def __init__(self, session: AsyncSession):
|
|
self.session = session
|
|
|
|
async def _serialize_rows(self, rows):
|
|
user_ids = set()
|
|
for row in rows:
|
|
user_ids |= {row.recruiter_id, row.candidate_id, row.created_by}
|
|
names = await Users.names_by_ids(self.session, user_ids)
|
|
return [serialize_reference_check(row, names) for row in rows]
|
|
|
|
async def create_form(self, payload, current_user):
|
|
if payload.get("record_overview") and "date_time" in payload["record_overview"]:
|
|
payload["record_overview"]["date_time"] = _aware(payload["record_overview"]["date_time"])
|
|
payload["created_by"] = _user_id(current_user)
|
|
row = await ReferenceCheck.insert_form(self.session, payload)
|
|
return (await self._serialize_rows([row]))[0]
|
|
|
|
async def update_form(self, form_id, payload, current_user):
|
|
_user_id(current_user)
|
|
row = await ReferenceCheck.get_form_by_id(self.session, form_id)
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="Reference check not found")
|
|
if not payload:
|
|
raise HTTPException(status_code=400, detail="No fields to update")
|
|
if payload.get("record_overview") and "date_time" in payload["record_overview"]:
|
|
payload["record_overview"]["date_time"] = _aware(payload["record_overview"]["date_time"])
|
|
updated = await ReferenceCheck.update_form(self.session, form_id, payload)
|
|
if not updated:
|
|
raise HTTPException(status_code=404, detail="Reference check not found")
|
|
return (await self._serialize_rows([updated]))[0]
|
|
|
|
async def get_form_by_id(self, form_id, candidate_id, current_user):
|
|
created_by = None if is_admin(current_user) else _user_id(current_user)
|
|
if form_id:
|
|
row = await ReferenceCheck.get_form_by_id(self.session, record_id=form_id, created_by=created_by)
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="Reference check not found")
|
|
return (await self._serialize_rows([row]))[0]
|
|
rows = await ReferenceCheck.get_form_by_id(
|
|
self.session, created_by=created_by, candidate_id=_as_uuid(candidate_id)
|
|
)
|
|
return await self._serialize_rows(rows)
|
|
|
|
async def delete_form(self, form_id, current_user):
|
|
_user_id(current_user)
|
|
row = await ReferenceCheck.soft_delete_form(self.session, form_id)
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="Reference check not found")
|
|
return {"id": str(row.id), "deleted": True}
|