Spaces:
Runtime error
Runtime error
File size: 5,116 Bytes
3e28a11 |
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 |
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
solicitation_no: str
solicitation_title: str
google_drive_id: Optional[str] = None
status: RFPStatus
created_at: datetime
updated_at: datetime
class ResponseRFP(BaseModel):
status: str
data: Optional[List[RFP]]
class REFPSingleResponse(BaseModel):
status: str
data: RFP
class RFPRequest(BaseModel):
solicitation_no: str
solicitation_title: str
google_drive_id: Optional[str] = None
status: Optional[RFPStatus] = None
class RFPUpdateRequest(BaseModel):
solicitation_no: Optional[str] = None
solicitation_title: Optional[str] = None
google_drive_id: Optional[str] = None
status: Optional[RFPStatus] = 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=REFPSingleResponse,
tags=["RFPs by ID"],
)
self.router.add_api_route(
"/rfps",
self.create_rfp,
methods=["POST"],
response_model=REFPSingleResponse,
tags=["RFPs"],
)
self.router.add_api_route(
"/rfps/{rfp_id}",
self.update_rfp,
methods=["PATCH"],
response_model=REFPSingleResponse,
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: UUID = Path(...)):
async with self.__rfp_service() as service:
try:
rfps = await service.get_rfps(rfp_id=rfp_id)
rfp = rfps[0]
return REFPSingleResponse(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 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 REFPSingleResponse(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: UUID = 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 REFPSingleResponse(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: UUID = 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.")
|