File size: 5,750 Bytes
3370983 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 |
"""
Database Query Schemas.
Flexible schemas for querying any table in the recruitment database.
"""
from datetime import datetime
from typing import Any, Optional
from uuid import UUID
from enum import Enum
from pydantic import BaseModel, Field
# ==================================================================================
# ENUMS
# ==================================================================================
class TableName(str, Enum):
"""Available tables for querying."""
candidates = "candidates"
cv_screening_results = "cv_screening_results"
voice_screening_results = "voice_screening_results"
interview_scheduling = "interview_scheduling"
final_decision = "final_decision"
class SortOrder(str, Enum):
"""Sort order options."""
asc = "asc"
desc = "desc"
# ==================================================================================
# REQUEST SCHEMAS
# ==================================================================================
class QueryRequest(BaseModel):
"""Flexible query request for any table."""
table: TableName = Field(..., description="Table to query")
# Filtering
filters: Optional[dict[str, Any]] = Field(
default=None,
description="Key-value filters (e.g., {'email': 'john@example.com', 'status': 'applied'})"
)
# Field selection
fields: Optional[list[str]] = Field(
default=None,
description="Specific fields to return. If None, returns all fields."
)
# Include related data
include_relations: Optional[bool] = Field(
default=False,
description="Include related tables (only for candidates table)"
)
# Pagination
limit: Optional[int] = Field(default=100, ge=1, le=1000, description="Max records to return")
offset: Optional[int] = Field(default=0, ge=0, description="Number of records to skip")
# Sorting
sort_by: Optional[str] = Field(default=None, description="Field to sort by")
sort_order: SortOrder = Field(default=SortOrder.desc, description="Sort order")
model_config = {
"json_schema_extra": {
"examples": [
{
"table": "candidates",
"filters": {"status": "applied"},
"fields": ["id", "full_name", "email", "status"],
"limit": 10
},
{
"table": "cv_screening_results",
"filters": {"overall_fit_score": {"$gte": 0.8}},
"sort_by": "overall_fit_score",
"sort_order": "desc"
}
]
}
}
# ==================================================================================
# RESPONSE SCHEMAS
# ==================================================================================
class CandidateResponse(BaseModel):
"""Candidate data response."""
id: UUID
full_name: str
email: str
phone_number: Optional[str] = None
cv_file_path: Optional[str] = None
parsed_cv_file_path: Optional[str] = None
status: str
created_at: datetime
updated_at: datetime
# Related data (populated when include_relations=True)
cv_screening_results: Optional[list[dict[str, Any]]] = None
voice_screening_results: Optional[list[dict[str, Any]]] = None
interview_scheduling: Optional[list[dict[str, Any]]] = None
final_decision: Optional[dict[str, Any]] = None
model_config = {"from_attributes": True}
class CVScreeningResponse(BaseModel):
"""CV Screening result response."""
id: UUID
candidate_id: UUID
job_title: Optional[str] = None
skills_match_score: Optional[float] = None
experience_match_score: Optional[float] = None
education_match_score: Optional[float] = None
overall_fit_score: Optional[float] = None
llm_feedback: Optional[str] = None
reasoning_trace: Optional[dict[str, Any]] = None
timestamp: datetime
model_config = {"from_attributes": True}
class VoiceScreeningResponse(BaseModel):
"""Voice Screening result response."""
id: UUID
candidate_id: UUID
call_sid: Optional[str] = None
transcript_text: Optional[str] = None
sentiment_score: Optional[float] = None
confidence_score: Optional[float] = None
communication_score: Optional[float] = None
llm_summary: Optional[str] = None
llm_judgment_json: Optional[dict[str, Any]] = None
audio_url: Optional[str] = None
timestamp: datetime
model_config = {"from_attributes": True}
class InterviewSchedulingResponse(BaseModel):
"""Interview scheduling response."""
id: UUID
candidate_id: UUID
calendar_event_id: Optional[str] = None
event_summary: Optional[str] = None
start_time: Optional[datetime] = None
end_time: Optional[datetime] = None
status: Optional[str] = None
timestamp: datetime
model_config = {"from_attributes": True}
class FinalDecisionResponse(BaseModel):
"""Final decision response."""
id: UUID
candidate_id: UUID
overall_score: Optional[float] = None
decision: Optional[str] = None
llm_rationale: Optional[str] = None
human_notes: Optional[str] = None
timestamp: datetime
model_config = {"from_attributes": True}
class QueryResponse(BaseModel):
"""Generic query response wrapper."""
success: bool
table: str
total_count: int
returned_count: int
offset: int
data: list[dict[str, Any]]
message: Optional[str] = None
class SingleRecordResponse(BaseModel):
"""Single record response."""
success: bool
table: str
data: Optional[dict[str, Any]] = None
message: Optional[str] = None
|