Spaces:
Sleeping
Sleeping
File size: 2,816 Bytes
0f24635 1dd2d1f 0f24635 ed0e747 0f24635 21e188e 0f24635 ed0e747 0f24635 |
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 |
from fastapi import APIRouter, Depends, HTTPException
from motor.motor_asyncio import AsyncIOMotorDatabase
from src.core.database import mongo_db
from src.models.mongo.interview_history_model import InterviewHistoryModel
from pydantic import BaseModel, Field
from typing import Optional, List, Dict, Any
from bson import ObjectId
router = APIRouter()
class InterviewHistoryCreate(BaseModel):
user_id: str
cv_id: str
job_offer_id: str
conversation: List[Dict[str, Any]]
start_time: str
end_time: Optional[str] = None
class InterviewHistoryUpdate(BaseModel):
conversation: Optional[List[Dict[str, Any]]] = None
job_offer_snapshot: Optional[Dict[str, Any]] = None
end_time: Optional[str] = None
class InterviewHistoryResponse(BaseModel):
id: str = Field(alias="_id")
user_id: str
cv_id: str
job_offer_id: str
conversation: List[Dict[str, Any]]
start_time: str
end_time: Optional[str] = None
class Config:
populate_by_name = True
json_encoders = {ObjectId: str}
@router.post("/interview-histories", response_model=InterviewHistoryResponse)
async def create_interview_history(history: InterviewHistoryCreate, db: AsyncIOMotorDatabase = Depends(lambda: mongo_db)):
history_entry = InterviewHistoryModel(**history.model_dump(by_alias=True))
history_id = await InterviewHistoryModel.create(db, InterviewHistoryModel.collection_name, history_entry.model_dump(exclude_unset=True))
history_entry.id = history_id
return history_entry
@router.get("/interview-histories/{history_id}", response_model=InterviewHistoryResponse)
async def get_interview_history_by_id(history_id: str, db: AsyncIOMotorDatabase = Depends(lambda: mongo_db)):
if not ObjectId.is_valid(history_id):
raise HTTPException(status_code=400, detail="Invalid ID format")
history = await InterviewHistoryModel.get(db, InterviewHistoryModel.collection_name, {"_id": ObjectId(history_id)})
if history is None:
raise HTTPException(status_code=404, detail="Interview history not found")
# Convert ObjectId to string to satisfy Pydantic response model
if "_id" in history:
history["_id"] = str(history["_id"])
return history
@router.put("/interview-histories/{history_id}", response_model=InterviewHistoryResponse)
async def update_interview_history(history_id: str, history: InterviewHistoryUpdate, db: AsyncIOMotorDatabase = Depends(lambda: mongo_db)):
if not ObjectId.is_valid(history_id):
raise HTTPException(status_code=400, detail="Invalid ID format")
await InterviewHistoryModel.update(db, InterviewHistoryModel.collection_name, {"_id": ObjectId(history_id)}, history.model_dump(exclude_unset=True))
return await get_interview_history_by_id(history_id, db)
|