Spaces:
Runtime error
Runtime error
File size: 5,741 Bytes
1f780d7 8e1c132 aa5bf7c 8e1c132 2aa3f0a 6c377a5 8e1c132 ef8c55b 8e1c132 aa5bf7c 7a511fb 2aa3f0a 9b0a6de 8e1c132 6c377a5 8e1c132 6c377a5 8e1c132 ef8c55b 8e1c132 7a511fb 2aa3f0a 9b0a6de 8e1c132 6c377a5 8e1c132 ef8c55b 8e1c132 aa5bf7c 7a511fb 2aa3f0a 9b0a6de 8e1c132 6c377a5 8e1c132 6c377a5 8e1c132 7a511fb 8e1c132 7a511fb 8e1c132 7a511fb 8e1c132 7a511fb 1f780d7 8e1c132 6c377a5 1f780d7 8e1c132 2aa3f0a 1f780d7 2aa3f0a 1f780d7 7a511fb 1f780d7 8e1c132 6c377a5 8e1c132 a17ec0b 8e1c132 6c377a5 1f780d7 8e1c132 7a511fb 8e1c132 6c377a5 1f780d7 8e1c132 6c377a5 |
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 |
from fastapi import Path
from fastapi import APIRouter, HTTPException, Query
from pydantic import BaseModel, Field, field_validator
from typing import List, Optional
from uuid import UUID
from src.config import logger
from src.services import RFPService
from datetime import datetime
from src.models import RFPStatus
class RFP(BaseModel):
id: UUID
rfp_number: str
name: str
received_proposals: int
evaluated_count: int
awaiting_evaluation: int
target_award_date: Optional[datetime] = None
ko_name: Optional[str] = None
award_type: Optional[str] = None
google_drive_id: Optional[str] = None
status: RFPStatus
comparative_weights_locked: Optional[bool] = False
created_at: datetime
updated_at: datetime
class ResponseRFP(BaseModel):
status: str
data: Optional[List[RFP]]
class RFPRequest(BaseModel):
rfp_number: str
name: str
received_proposals: int
evaluated_count: int
target_award_date: Optional[datetime] = None
ko_name: Optional[str] = None
award_type: Optional[str] = None
google_drive_id: Optional[str] = None
status: RFPStatus
comparative_weights_locked: Optional[bool] = False
awaiting_evaluation: int
class RFPUpdateRequest(BaseModel):
rfp_number: Optional[str] = None
name: Optional[str] = None
received_proposals: Optional[int] = None
evaluated_count: Optional[int] = None
target_award_date: Optional[datetime] = None
ko_name: Optional[str] = None
award_type: Optional[str] = None
google_drive_id: Optional[str] = None
status: Optional[RFPStatus] = None
comparative_weights_locked: Optional[bool] = None
awaiting_evaluation: Optional[int] = None
class RFPDeleteResponse(BaseModel):
status: str
class RFPController:
def __init__(self):
self.__rfp_service = RFPService
self.router = APIRouter()
self.router.add_api_route(
"/rfps",
self.get_rfps,
methods=["GET"],
response_model=ResponseRFP,
tags=["RFPs"],
)
self.router.add_api_route(
"/rfps/{rfp_id}",
self.get_rfps_by_id,
methods=["GET"],
response_model=ResponseRFP,
tags=["RFPs by ID"],
)
self.router.add_api_route(
"/rfps",
self.create_rfp,
methods=["POST"],
response_model=ResponseRFP,
tags=["RFPs"],
)
self.router.add_api_route(
"/rfps/{rfp_id}",
self.update_rfp,
methods=["PUT"],
response_model=ResponseRFP,
tags=["RFPs by ID"],
)
self.router.add_api_route(
"/rfps/{rfp_id}",
self.delete_rfp,
methods=["DELETE"],
response_model=RFPDeleteResponse,
tags=["RFPs by ID"],
)
async def get_rfps(self, status: Optional[RFPStatus] = Query(None)):
async with self.__rfp_service() as service:
try:
rfps = await service.get_rfps(status=status)
return ResponseRFP(status="success", data=[RFP(**rfp) for rfp in rfps])
except HTTPException as e:
logger.warning(e)
raise e
except Exception as e:
logger.exception(e)
raise HTTPException(status_code=500, detail="Failed to retrieve RFPs.")
async def get_rfps_by_id(self, rfp_id: str = Path(...)):
async with self.__rfp_service() as service:
try:
rfps = await service.get_rfps(rfp_id=rfp_id)
return ResponseRFP(status="success", data=[RFP(**rfp) for rfp in rfps])
except HTTPException as e:
logger.warning(e)
raise e
except Exception as e:
logger.exception(e)
raise HTTPException(status_code=500, detail="Failed to retrieve RFPs.")
async def create_rfp(self, rfp: RFPRequest):
async with self.__rfp_service() as service:
try:
rfp = await service.create_rfp(rfp.model_dump(exclude_unset=True))
return ResponseRFP(status="success", data=[RFP(**rfp)])
except HTTPException as e:
logger.warning(e)
raise e
except Exception as e:
logger.exception(e)
raise HTTPException(status_code=500, detail="Failed to create RFP.")
async def update_rfp(self, rfp: RFPUpdateRequest, rfp_id: str = Path(...)):
async with self.__rfp_service() as service:
try:
rfp = await service.update_rfp(
rfp_id=rfp_id, rfp=rfp.model_dump(exclude_unset=True)
)
return ResponseRFP(status="success", data=[RFP(**rfp)])
except HTTPException as e:
logger.warning(e)
raise e
except Exception as e:
logger.exception(e)
raise HTTPException(status_code=500, detail="Failed to update RFP.")
async def delete_rfp(self, rfp_id: str = Path(...)):
async with self.__rfp_service() as service:
try:
status = await service.delete_rfp(rfp_id)
if not status:
raise HTTPException(status_code=404, detail="RFP not found.")
return RFPDeleteResponse(status="success")
except HTTPException as e:
logger.warning(e)
raise e
except Exception as e:
logger.exception(e)
raise HTTPException(status_code=500, detail="Failed to delete RFP.")
|