Spaces:
Runtime error
Runtime error
| from fastapi import APIRouter, HTTPException, Query, Path | |
| from pydantic import BaseModel, field_validator | |
| from typing import List, Optional | |
| from uuid import UUID | |
| from datetime import datetime | |
| from src.config import logger | |
| from src.services import ProposalService | |
| from src.models import ProposalStatus, GateCriteria | |
| class Proposal(BaseModel): | |
| id: UUID | |
| rfp_id: UUID | |
| name: str | |
| tep: str | |
| gate_criteria: GateCriteria | |
| status: ProposalStatus | |
| ai_score: Optional[float] = None | |
| final_score: Optional[float] = None | |
| created_at: datetime | |
| updated_at: datetime | |
| def round_to_two_decimals(cls, v): | |
| if v is not None: | |
| return round(v, 2) | |
| return v | |
| class ProposalRequest(BaseModel): | |
| rfp_id: UUID | |
| name: str | |
| tep: str | |
| gate_criteria: GateCriteria | |
| status: ProposalStatus | |
| ai_score: Optional[float] = None | |
| final_score: Optional[float] = None | |
| class CreateProposalRequest(BaseModel): | |
| name: str | |
| tep: str | |
| gate_criteria: GateCriteria | |
| status: ProposalStatus | |
| ai_score: Optional[float] = None | |
| final_score: Optional[float] = None | |
| class ProposalUpdateRequest(BaseModel): | |
| rfp_id: Optional[UUID] = None | |
| name: Optional[str] = None | |
| tep: Optional[str] = None | |
| gate_criteria: Optional[GateCriteria] = None | |
| status: Optional[ProposalStatus] = None | |
| ai_score: Optional[float] = None | |
| final_score: Optional[float] = None | |
| class UpdateProposalRequest(BaseModel): | |
| name: Optional[str] = None | |
| tep: Optional[str] = None | |
| gate_criteria: Optional[GateCriteria] = None | |
| status: Optional[ProposalStatus] = None | |
| ai_score: Optional[float] = None | |
| final_score: Optional[float] = None | |
| class ProposalDeleteResponse(BaseModel): | |
| status: str | |
| class ResponseProposal(BaseModel): | |
| status: str | |
| data: Optional[List[Proposal]] | |
| class ProposalController: | |
| def __init__(self): | |
| self.__proposal_service = ProposalService | |
| self.router = APIRouter() | |
| # self.router.add_api_route( | |
| # "/proposals", | |
| # self.get_proposals, | |
| # methods=["GET"], | |
| # response_model=ResponseProposal, | |
| # tags=["Proposals"], | |
| # ) | |
| # self.router.add_api_route( | |
| # "/proposals/{proposal_id}", | |
| # self.get_proposals_by_id, | |
| # methods=["GET"], | |
| # response_model=ResponseProposal, | |
| # tags=["Proposals by ID"], | |
| # ) | |
| self.router.add_api_route( | |
| "/rfps/{rfp_id}/proposals", | |
| self.get_proposals_by_rfp_id, | |
| methods=["GET"], | |
| response_model=ResponseProposal, | |
| tags=["Proposals by RFP ID"], | |
| ) | |
| self.router.add_api_route( | |
| "/rfps/{rfp_id}/proposals/{proposal_id}", | |
| self.get_proposals_by_proposal_and_rfp_id, | |
| methods=["GET"], | |
| response_model=ResponseProposal, | |
| tags=["Proposals by Proposal and RFP ID"], | |
| ) | |
| # self.router.add_api_route( | |
| # "/proposals", | |
| # self.create_proposal, | |
| # methods=["POST"], | |
| # response_model=ResponseProposal, | |
| # tags=["Proposals"], | |
| # ) | |
| self.router.add_api_route( | |
| "/rfps/{rfp_id}/proposals", | |
| self.create_proposal_by_rfp_id, | |
| methods=["POST"], | |
| response_model=ResponseProposal, | |
| tags=["Proposals by RFP ID"], | |
| ) | |
| # self.router.add_api_route( | |
| # "/proposals/{proposal_id}", | |
| # self.update_proposal, | |
| # methods=["PUT"], | |
| # response_model=ResponseProposal, | |
| # tags=["Proposals by ID"], | |
| # ) | |
| self.router.add_api_route( | |
| "/rfps/{rfp_id}/proposals/{proposal_id}", | |
| self.update_proposal_by_proposal_and_rfp_id, | |
| methods=["PUT"], | |
| response_model=ResponseProposal, | |
| tags=["Proposals by Proposal and RFP ID"], | |
| ) | |
| # self.router.add_api_route( | |
| # "/proposals/{proposal_id}", | |
| # self.delete_proposal, | |
| # methods=["DELETE"], | |
| # response_model=ProposalDeleteResponse, | |
| # tags=["Proposals by ID"], | |
| # ) | |
| self.router.add_api_route( | |
| "/rfps/{rfp_id}/proposals", | |
| self.delete_proposals_by_rfp_id, | |
| methods=["DeLETE"], | |
| response_model=ProposalDeleteResponse, | |
| tags=["Proposals by RFP ID"], | |
| ) | |
| self.router.add_api_route( | |
| "/rfps/{rfp_id}/proposals/{proposal_id}", | |
| self.delete_proposals_by_proposal_and_rfp_id, | |
| methods=["DeLETE"], | |
| response_model=ProposalDeleteResponse, | |
| tags=["Proposals by Proposal and RFP ID"], | |
| ) | |
| async def get_proposals(self): | |
| async with self.__proposal_service() as service: | |
| try: | |
| proposals = await service.get_proposals() | |
| return ResponseProposal( | |
| status="success", data=[Proposal(**p) for p in proposals] | |
| ) | |
| except Exception as e: | |
| logger.exception(e) | |
| raise HTTPException( | |
| status_code=500, detail="Failed to retrieve proposals." | |
| ) | |
| async def get_proposals_by_id( | |
| self, | |
| proposal_id: str = Path(...), | |
| ): | |
| async with self.__proposal_service() as service: | |
| try: | |
| proposals = await service.get_proposals(proposal_id=proposal_id) | |
| return ResponseProposal( | |
| status="success", data=[Proposal(**p) for p in proposals] | |
| ) | |
| except Exception as e: | |
| logger.exception(e) | |
| raise HTTPException( | |
| status_code=500, detail="Failed to retrieve proposals." | |
| ) | |
| async def get_proposals_by_rfp_id( | |
| self, | |
| rfp_id: str = Path(...), | |
| ): | |
| async with self.__proposal_service() as service: | |
| try: | |
| proposals = await service.get_proposals(rfp_id=rfp_id) | |
| return ResponseProposal( | |
| status="success", data=[Proposal(**p) for p in proposals] | |
| ) | |
| except Exception as e: | |
| logger.exception(e) | |
| raise HTTPException( | |
| status_code=500, detail="Failed to retrieve proposals." | |
| ) | |
| async def get_proposals_by_proposal_and_rfp_id( | |
| self, | |
| rfp_id: str = Path(...), | |
| proposal_id: str = Path(...), | |
| ): | |
| async with self.__proposal_service() as service: | |
| try: | |
| proposals = await service.get_proposals( | |
| rfp_id=rfp_id, proposal_id=proposal_id | |
| ) | |
| return ResponseProposal( | |
| status="success", data=[Proposal(**p) for p in proposals] | |
| ) | |
| except Exception as e: | |
| logger.exception(e) | |
| raise HTTPException( | |
| status_code=500, detail="Failed to retrieve proposals." | |
| ) | |
| async def create_proposal(self, proposal: ProposalRequest): | |
| async with self.__proposal_service() as service: | |
| try: | |
| new_proposal = await service.create_proposal( | |
| proposal.model_dump(mode="json") | |
| ) | |
| return ResponseProposal( | |
| status="success", data=[Proposal(**new_proposal)] | |
| ) | |
| except Exception as e: | |
| logger.exception(e) | |
| raise HTTPException( | |
| status_code=500, detail="Failed to create proposal." | |
| ) | |
| async def create_proposal_by_rfp_id( | |
| self, proposal: CreateProposalRequest, rfp_id: str = Path(...) | |
| ): | |
| async with self.__proposal_service() as service: | |
| try: | |
| data = proposal.model_dump(exclude_unset=True, mode="json") | |
| data["rfp_id"] = rfp_id | |
| new_proposal = await service.create_proposal(proposal=data) | |
| return ResponseProposal( | |
| status="success", data=[Proposal(**new_proposal)] | |
| ) | |
| except Exception as e: | |
| logger.exception(e) | |
| raise HTTPException( | |
| status_code=500, detail="Failed to create proposal." | |
| ) | |
| async def update_proposal( | |
| self, proposal: ProposalUpdateRequest, proposal_id: str = Path(...) | |
| ): | |
| async with self.__proposal_service() as service: | |
| try: | |
| updated_proposal = await service.update_proposal( | |
| proposal_id=proposal_id, | |
| proposal=proposal.model_dump(exclude_unset=True), | |
| ) | |
| return ResponseProposal( | |
| status="success", data=[Proposal(**updated_proposal)] | |
| ) | |
| except Exception as e: | |
| logger.exception(e) | |
| raise HTTPException( | |
| status_code=500, detail="Failed to update proposal." | |
| ) | |
| async def update_proposal_by_proposal_and_rfp_id( | |
| self, | |
| proposal: UpdateProposalRequest, | |
| rfp_id: str = Path(...), | |
| proposal_id: str = Path(...), | |
| ): | |
| async with self.__proposal_service() as service: | |
| try: | |
| data = proposal.model_dump(exclude_unset=True, mode="json") | |
| data["rfp_id"] = rfp_id | |
| updated_proposal = await service.update_proposal( | |
| proposal_id=proposal_id, | |
| proposal=proposal.model_dump(exclude_unset=True), | |
| ) | |
| return ResponseProposal( | |
| status="success", data=[Proposal(**updated_proposal)] | |
| ) | |
| except Exception as e: | |
| logger.exception(e) | |
| raise HTTPException( | |
| status_code=500, detail="Failed to update proposal." | |
| ) | |
| async def delete_proposal(self, proposal_id: str = Path(...)): | |
| async with self.__proposal_service() as service: | |
| try: | |
| status = await service.delete_proposal(proposal_id=proposal_id) | |
| if not status: | |
| raise HTTPException(status_code=404, detail="Proposal not found.") | |
| return ProposalDeleteResponse(status="success") | |
| except Exception as e: | |
| logger.exception(e) | |
| raise HTTPException( | |
| status_code=500, detail="Failed to delete proposal." | |
| ) | |
| async def delete_proposals_by_rfp_id(self, rfp_id: str = Path(...)): | |
| async with self.__proposal_service() as service: | |
| try: | |
| status = await service.delete_proposal(rfp_id=rfp_id) | |
| if not status: | |
| raise HTTPException(status_code=404, detail="Proposal not found.") | |
| return ProposalDeleteResponse(status="success") | |
| except Exception as e: | |
| logger.exception(e) | |
| raise HTTPException( | |
| status_code=500, detail="Failed to delete proposal." | |
| ) | |
| async def delete_proposals_by_proposal_and_rfp_id( | |
| self, rfp_id: str = Path(...), proposal_id: str = Path(...) | |
| ): | |
| async with self.__proposal_service() as service: | |
| try: | |
| status = await service.delete_proposal( | |
| rfp_id=rfp_id, proposal_id=proposal_id | |
| ) | |
| if not status: | |
| raise HTTPException(status_code=404, detail="Proposal not found.") | |
| return ProposalDeleteResponse(status="success") | |
| except Exception as e: | |
| logger.exception(e) | |
| raise HTTPException( | |
| status_code=500, detail="Failed to delete proposal." | |
| ) | |