154 lines
8.2 KiB
Python
154 lines
8.2 KiB
Python
import uuid
|
|
from datetime import datetime, date as Date, timezone
|
|
from typing import Optional
|
|
|
|
from sqlalchemy import DateTime
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlmodel import Field, SQLModel, select
|
|
|
|
|
|
def _now() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
class ReferenceCheck(SQLModel, table=True):
|
|
"""Annexure G - Employment Reference Check Form. One row per referee, so a
|
|
candidate can hold several. recruiter_id / candidate_id are both users.id."""
|
|
|
|
__tablename__ = "Reference_Check"
|
|
id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
|
|
|
|
recruiter_id: Optional[uuid.UUID] = Field(default=None, foreign_key="users.id", index=True)
|
|
candidate_id: Optional[uuid.UUID] = Field(default=None, foreign_key="users.id", index=True)
|
|
|
|
position_applied: Optional[str] = None
|
|
referee_name: Optional[str] = None
|
|
date_time: Optional[datetime] = Field(default=None, sa_type=DateTime(timezone=True))
|
|
referee_title: Optional[str] = None
|
|
referee_organization: Optional[str] = None
|
|
relationship_duration: Optional[str] = None
|
|
|
|
scope_of_relationship: Optional[str] = None
|
|
work_approach: Optional[str] = None
|
|
leadership: Optional[str] = None
|
|
communication: Optional[str] = None
|
|
agility: Optional[str] = None
|
|
|
|
signoff_referee_name: Optional[str] = None
|
|
signoff_designation: Optional[str] = None
|
|
signoff_date: Optional[Date] = None
|
|
|
|
created_by: Optional[uuid.UUID] = Field(foreign_key="users.id")
|
|
created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
|
|
updated_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
|
|
is_deleted: bool = Field(default=False)
|
|
|
|
@classmethod
|
|
async def get_form_by_id(cls, session: AsyncSession, record_id=None, created_by=None, candidate_id=None):
|
|
qry = select(cls).where(cls.is_deleted == False) # noqa: E712
|
|
if created_by is not None:
|
|
qry = qry.where(cls.created_by == created_by)
|
|
if candidate_id is not None:
|
|
qry = qry.where(cls.candidate_id == candidate_id)
|
|
if record_id not in (None, ""):
|
|
try:
|
|
uid = uuid.UUID(str(record_id))
|
|
except (TypeError, ValueError):
|
|
return None
|
|
qry = qry.where(cls.id == uid)
|
|
result = await session.execute(qry)
|
|
return result.scalars().first()
|
|
result = await session.execute(qry.order_by(cls.created_at.desc(), cls.id.desc()))
|
|
return list(result.scalars().all())
|
|
|
|
@classmethod
|
|
async def insert_form(cls, session: AsyncSession, fields: dict):
|
|
overview = fields.get("record_overview") if fields.get("record_overview") else {}
|
|
feedback = fields.get("professional_feedback") if fields.get("professional_feedback") else {}
|
|
signoff = fields.get("referee_signoff") if fields.get("referee_signoff") else {}
|
|
row = cls(
|
|
recruiter_id=fields.get("recruiter_id") if fields.get("recruiter_id") else None,
|
|
candidate_id=fields.get("candidate_id") if fields.get("candidate_id") else None,
|
|
position_applied=overview.get("position_applied") if overview.get("position_applied") else None,
|
|
referee_name=overview.get("referee_name") if overview.get("referee_name") else None,
|
|
date_time=overview.get("date_time") if overview.get("date_time") else None,
|
|
referee_title=overview.get("referee_title") if overview.get("referee_title") else None,
|
|
referee_organization=overview.get("referee_organization") if overview.get("referee_organization") else None,
|
|
relationship_duration=overview.get("relationship_duration") if overview.get("relationship_duration") else None,
|
|
scope_of_relationship=feedback.get("scope_of_relationship") if feedback.get("scope_of_relationship") else None,
|
|
work_approach=feedback.get("work_approach") if feedback.get("work_approach") else None,
|
|
leadership=feedback.get("leadership") if feedback.get("leadership") else None,
|
|
communication=feedback.get("communication") if feedback.get("communication") else None,
|
|
agility=feedback.get("agility") if feedback.get("agility") else None,
|
|
signoff_referee_name=signoff.get("referee_name") if signoff.get("referee_name") else None,
|
|
signoff_designation=signoff.get("designation") if signoff.get("designation") else None,
|
|
signoff_date=signoff.get("signed_date") if signoff.get("signed_date") else None,
|
|
created_by=fields.get("created_by") if fields.get("created_by") else None,
|
|
)
|
|
session.add(row)
|
|
await session.commit()
|
|
return await cls.get_form_by_id(session, row.id)
|
|
|
|
@classmethod
|
|
async def update_form(cls, session: AsyncSession, record_id, fields: dict):
|
|
row = await cls.get_form_by_id(session, record_id)
|
|
if not row:
|
|
return None
|
|
if "recruiter_id" in fields:
|
|
row.recruiter_id = fields.get("recruiter_id") if fields.get("recruiter_id") else None
|
|
if "candidate_id" in fields:
|
|
row.candidate_id = fields.get("candidate_id") if fields.get("candidate_id") else None
|
|
if "record_overview" in fields:
|
|
overview = fields.get("record_overview") if fields.get("record_overview") else {}
|
|
if "position_applied" in overview:
|
|
row.position_applied = overview.get("position_applied") if overview.get("position_applied") else None
|
|
if "referee_name" in overview:
|
|
row.referee_name = overview.get("referee_name") if overview.get("referee_name") else None
|
|
if "date_time" in overview:
|
|
row.date_time = overview.get("date_time") if overview.get("date_time") else None
|
|
if "referee_title" in overview:
|
|
row.referee_title = overview.get("referee_title") if overview.get("referee_title") else None
|
|
if "referee_organization" in overview:
|
|
row.referee_organization = overview.get("referee_organization") if overview.get("referee_organization") else None
|
|
if "relationship_duration" in overview:
|
|
row.relationship_duration = overview.get("relationship_duration") if overview.get("relationship_duration") else None
|
|
if "professional_feedback" in fields:
|
|
feedback = fields.get("professional_feedback") if fields.get("professional_feedback") else {}
|
|
if "scope_of_relationship" in feedback:
|
|
row.scope_of_relationship = feedback.get("scope_of_relationship") if feedback.get("scope_of_relationship") else None
|
|
if "work_approach" in feedback:
|
|
row.work_approach = feedback.get("work_approach") if feedback.get("work_approach") else None
|
|
if "leadership" in feedback:
|
|
row.leadership = feedback.get("leadership") if feedback.get("leadership") else None
|
|
if "communication" in feedback:
|
|
row.communication = feedback.get("communication") if feedback.get("communication") else None
|
|
if "agility" in feedback:
|
|
row.agility = feedback.get("agility") if feedback.get("agility") else None
|
|
if "referee_signoff" in fields:
|
|
signoff = fields.get("referee_signoff") if fields.get("referee_signoff") else {}
|
|
if "referee_name" in signoff:
|
|
row.signoff_referee_name = signoff.get("referee_name") if signoff.get("referee_name") else None
|
|
if "designation" in signoff:
|
|
row.signoff_designation = signoff.get("designation") if signoff.get("designation") else None
|
|
if "signed_date" in signoff:
|
|
row.signoff_date = signoff.get("signed_date") if signoff.get("signed_date") else None
|
|
row.updated_at = _now()
|
|
session.add(row)
|
|
await session.commit()
|
|
await session.refresh(row)
|
|
return row
|
|
|
|
@classmethod
|
|
async def soft_delete_form(cls, session: AsyncSession, record_id):
|
|
row = await cls.get_form_by_id(session, record_id)
|
|
if not row:
|
|
return None
|
|
row.is_deleted = True
|
|
row.updated_at = _now()
|
|
session.add(row)
|
|
await session.commit()
|
|
return row
|
|
|
|
|
|
import users.models as _users_models # noqa: E402, F401
|