import uuid from datetime import datetime, date as Date, timezone from typing import Optional from sqlalchemy import DateTime, Enum as SAEnum from sqlalchemy.ext.asyncio import AsyncSession from sqlmodel import Field, SQLModel, select from verification_check.enums import ReHireEligibility def _now() -> datetime: return datetime.now(timezone.utc) class VerificationCheck(SQLModel, table=True): """Annexure E - Employment Verification Check Form. One row per referee call, so a candidate can hold several. recruiter_id / candidate_id are both users.id.""" __tablename__ = "Verification_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 department_id: Optional[uuid.UUID] = Field(default=None, foreign_key="departments.id") department: Optional[str] = None date_time: Optional[datetime] = Field(default=None, sa_type=DateTime(timezone=True)) confirmed_job_title: Optional[str] = None confirmed_from: Optional[Date] = None confirmed_to: Optional[Date] = None reason_for_leaving: Optional[str] = None technical_knowledge_rating: Optional[int] = None technical_knowledge_notes: Optional[str] = None reliability_rating: Optional[int] = None reliability_notes: Optional[str] = None communication_rating: Optional[int] = None communication_notes: Optional[str] = None problem_solving_rating: Optional[int] = None problem_solving_notes: Optional[str] = None strengths: Optional[str] = None areas_for_growth: Optional[str] = None rehire_eligibility: Optional[ReHireEligibility] = Field( default=None, sa_type=SAEnum( ReHireEligibility, name="rehireeligibility", schema="app", native_enum=True, values_callable=lambda enum: [member.value for member in enum], ), ) rehire_comments: Optional[str] = None referee_full_name: Optional[str] = None referee_current_title: Optional[str] = None referee_company: Optional[str] = None referee_phone: Optional[str] = None referee_email: Optional[str] = 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 {} employment = fields.get("employment_verification") if fields.get("employment_verification") else {} competency = fields.get("competency_evaluation") if fields.get("competency_evaluation") else {} technical = competency.get("technical_knowledge") if competency.get("technical_knowledge") else {} reliability = competency.get("reliability") if competency.get("reliability") else {} communication = competency.get("communication") if competency.get("communication") else {} problem_solving = competency.get("problem_solving") if competency.get("problem_solving") else {} qualitative = fields.get("qualitative_evaluation") if fields.get("qualitative_evaluation") else {} referee = fields.get("referee") if fields.get("referee") 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, department_id=overview.get("department_id") if overview.get("department_id") else None, department=overview.get("department") if overview.get("department") else None, date_time=overview.get("date_time") if overview.get("date_time") else None, confirmed_job_title=employment.get("confirmed_job_title") if employment.get("confirmed_job_title") else None, confirmed_from=employment.get("confirmed_from") if employment.get("confirmed_from") else None, confirmed_to=employment.get("confirmed_to") if employment.get("confirmed_to") else None, reason_for_leaving=employment.get("reason_for_leaving") if employment.get("reason_for_leaving") else None, technical_knowledge_rating=int(technical.get("rating")) if technical.get("rating") else None, technical_knowledge_notes=technical.get("notes") if technical.get("notes") else None, reliability_rating=int(reliability.get("rating")) if reliability.get("rating") else None, reliability_notes=reliability.get("notes") if reliability.get("notes") else None, communication_rating=int(communication.get("rating")) if communication.get("rating") else None, communication_notes=communication.get("notes") if communication.get("notes") else None, problem_solving_rating=int(problem_solving.get("rating")) if problem_solving.get("rating") else None, problem_solving_notes=problem_solving.get("notes") if problem_solving.get("notes") else None, strengths=qualitative.get("strengths") if qualitative.get("strengths") else None, areas_for_growth=qualitative.get("areas_for_growth") if qualitative.get("areas_for_growth") else None, rehire_eligibility=ReHireEligibility(qualitative.get("rehire_eligibility")) if qualitative.get("rehire_eligibility") else None, rehire_comments=qualitative.get("rehire_comments") if qualitative.get("rehire_comments") else None, referee_full_name=referee.get("full_name") if referee.get("full_name") else None, referee_current_title=referee.get("current_title") if referee.get("current_title") else None, referee_company=referee.get("company") if referee.get("company") else None, referee_phone=referee.get("phone") if referee.get("phone") else None, referee_email=referee.get("email") if referee.get("email") 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 "department_id" in overview: row.department_id = overview.get("department_id") if overview.get("department_id") else None if "department" in overview: row.department = overview.get("department") if overview.get("department") else None if "date_time" in overview: row.date_time = overview.get("date_time") if overview.get("date_time") else None if "employment_verification" in fields: employment = fields.get("employment_verification") if fields.get("employment_verification") else {} if "confirmed_job_title" in employment: row.confirmed_job_title = employment.get("confirmed_job_title") if employment.get("confirmed_job_title") else None if "confirmed_from" in employment: row.confirmed_from = employment.get("confirmed_from") if employment.get("confirmed_from") else None if "confirmed_to" in employment: row.confirmed_to = employment.get("confirmed_to") if employment.get("confirmed_to") else None if "reason_for_leaving" in employment: row.reason_for_leaving = employment.get("reason_for_leaving") if employment.get("reason_for_leaving") else None if "competency_evaluation" in fields: competency = fields.get("competency_evaluation") if fields.get("competency_evaluation") else {} if "technical_knowledge" in competency: technical = competency.get("technical_knowledge") if competency.get("technical_knowledge") else {} if "rating" in technical: row.technical_knowledge_rating = int(technical.get("rating")) if technical.get("rating") else None if "notes" in technical: row.technical_knowledge_notes = technical.get("notes") if technical.get("notes") else None if "reliability" in competency: reliability = competency.get("reliability") if competency.get("reliability") else {} if "rating" in reliability: row.reliability_rating = int(reliability.get("rating")) if reliability.get("rating") else None if "notes" in reliability: row.reliability_notes = reliability.get("notes") if reliability.get("notes") else None if "communication" in competency: communication = competency.get("communication") if competency.get("communication") else {} if "rating" in communication: row.communication_rating = int(communication.get("rating")) if communication.get("rating") else None if "notes" in communication: row.communication_notes = communication.get("notes") if communication.get("notes") else None if "problem_solving" in competency: problem_solving = competency.get("problem_solving") if competency.get("problem_solving") else {} if "rating" in problem_solving: row.problem_solving_rating = int(problem_solving.get("rating")) if problem_solving.get("rating") else None if "notes" in problem_solving: row.problem_solving_notes = problem_solving.get("notes") if problem_solving.get("notes") else None if "qualitative_evaluation" in fields: qualitative = fields.get("qualitative_evaluation") if fields.get("qualitative_evaluation") else {} if "strengths" in qualitative: row.strengths = qualitative.get("strengths") if qualitative.get("strengths") else None if "areas_for_growth" in qualitative: row.areas_for_growth = qualitative.get("areas_for_growth") if qualitative.get("areas_for_growth") else None if "rehire_eligibility" in qualitative: row.rehire_eligibility = ReHireEligibility(qualitative.get("rehire_eligibility")) if qualitative.get("rehire_eligibility") else None if "rehire_comments" in qualitative: row.rehire_comments = qualitative.get("rehire_comments") if qualitative.get("rehire_comments") else None if "referee" in fields: referee = fields.get("referee") if fields.get("referee") else {} if "full_name" in referee: row.referee_full_name = referee.get("full_name") if referee.get("full_name") else None if "current_title" in referee: row.referee_current_title = referee.get("current_title") if referee.get("current_title") else None if "company" in referee: row.referee_company = referee.get("company") if referee.get("company") else None if "phone" in referee: row.referee_phone = referee.get("phone") if referee.get("phone") else None if "email" in referee: row.referee_email = referee.get("email") if referee.get("email") 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 import department.models as _department_models # noqa: E402, F401