Spaces:
Sleeping
Sleeping
File size: 2,344 Bytes
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 |
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
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)):
history = await InterviewHistoryModel.get(db, InterviewHistoryModel.collection_name, {"_id": history_id})
if history is None:
raise HTTPException(status_code=404, detail="Interview history not found")
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)):
await InterviewHistoryModel.update(db, InterviewHistoryModel.collection_name, {"_id": history_id}, history.model_dump(exclude_unset=True))
return await get_interview_history_by_id(history_id, db)
|