Spaces:
Runtime error
Runtime error
| from fastapi import File, Path, Response, UploadFile | |
| from fastapi.responses import FileResponse | |
| 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 SOWService | |
| from datetime import datetime | |
| # from src.models import SOW | |
| class SOW(BaseModel): | |
| id: UUID | |
| # rfp_id: UUID | |
| requirement: Optional[str] = None | |
| additional_info: Optional[str] = None | |
| sow_generated_text: Optional[str] = None | |
| created_at: datetime | |
| updated_at: datetime | |
| class ResponseSOW(BaseModel): | |
| status: str | |
| data: Optional[List[SOW]] | |
| class ResponseSOWSingle(BaseModel): | |
| status: str | |
| data: SOW | |
| class SOWRequest(BaseModel): | |
| requirement: Optional[str] = None | |
| additional_info: Optional[str] = None | |
| sow_generated_text: Optional[str] = None | |
| class SOWUpdateRequest(BaseModel): | |
| requirement: Optional[str] = None | |
| additional_info: Optional[str] = None | |
| sow_generated_text: Optional[str] = None | |
| class DeleteResponse(BaseModel): | |
| status: str | |
| class SOWController: | |
| def __init__(self): | |
| self.__sow_service = SOWService | |
| self.router = APIRouter() | |
| self.router.add_api_route( | |
| "/sows", | |
| self.get_sows, | |
| methods=["GET"], | |
| response_model=ResponseSOW, | |
| tags=["SOWs"], | |
| ) | |
| self.router.add_api_route( | |
| "/sows/{sow_id}", | |
| self.get_sows_by_id, | |
| methods=["GET"], | |
| response_model=ResponseSOWSingle, | |
| tags=["SOWs by SOW ID"], | |
| ) | |
| self.router.add_api_route( | |
| "/sows", | |
| self.create_sow, | |
| methods=["POST"], | |
| response_model=ResponseSOWSingle, | |
| tags=["SOWs"], | |
| ) | |
| self.router.add_api_route( | |
| "/sows/{sow_id}", | |
| self.update_sow, | |
| methods=["PATCH"], | |
| response_model=ResponseSOWSingle, | |
| tags=["SOWs by SOW ID"], | |
| ) | |
| self.router.add_api_route( | |
| "/sows/{sow_id}", | |
| self.delete_sow, | |
| methods=["DELETE"], | |
| response_model=DeleteResponse, | |
| tags=["SOWs by SOW ID"], | |
| ) | |
| self.router.add_api_route( | |
| "/sows/{sow_id}/generate", | |
| self.generate_sow, | |
| methods=["GET"], | |
| response_model=ResponseSOWSingle, | |
| tags=["Generate SOW"], | |
| ) | |
| self.router.add_api_route( | |
| "/sows/{sow_id}/download", | |
| self.download_sow, | |
| methods=["GET"], | |
| response_model=None, | |
| tags=["Download SOW"], | |
| response_class=FileResponse, | |
| ) | |
| self.router.add_api_route( | |
| "/sows/upload", | |
| self.upload_sow, | |
| methods=["POST"], | |
| response_model=ResponseSOWSingle, | |
| tags=["Upload Requirement File"], | |
| ) | |
| # self.router.add_api_route( | |
| # "/sows/{sow_id}/upload", | |
| # self.upload_sow_by_id, | |
| # methods=["POST"], | |
| # response_model=ResponseSOWSingle, | |
| # tags=["Upload Requirement File"], | |
| # ) | |
| async def get_sows(self): | |
| async with self.__sow_service() as service: | |
| try: | |
| sows = await service.get_sows() | |
| return ResponseSOW(status="success", data=[SOW(**sow) for sow in sows]) | |
| 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 SOWs.") | |
| async def get_sows_by_id(self, sow_id: UUID = Path(...)): | |
| async with self.__sow_service() as service: | |
| try: | |
| sows = await service.get_sows(sow_id=sow_id) | |
| sow = sows[0] | |
| return ResponseSOWSingle(status="success", data=SOW(**sow)) | |
| 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 SOWs.") | |
| async def create_sow(self, sow: SOWRequest): | |
| async with self.__sow_service() as service: | |
| try: | |
| sow = await service.create_sow( | |
| sow=sow.model_dump(mode="json", exclude_unset=True) | |
| ) | |
| return ResponseSOWSingle(status="success", data=SOW(**sow)) | |
| 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 SOW.") | |
| async def update_sow(self, sow: SOWUpdateRequest, sow_id: UUID = Path(...)): | |
| async with self.__sow_service() as service: | |
| try: | |
| sow = await service.update_sow( | |
| sow_id=sow_id, | |
| sow=sow.model_dump(mode="json", exclude_unset=True), | |
| ) | |
| return ResponseSOWSingle(status="success", data=SOW(**sow)) | |
| 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 SOW.") | |
| async def delete_sow(self, sow_id: UUID = Path(...)): | |
| async with self.__sow_service() as service: | |
| try: | |
| status = await service.delete_sow(sow_id=sow_id) | |
| if not status: | |
| raise HTTPException(status_code=404, detail="SOW not found.") | |
| return DeleteResponse(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 SOW.") | |
| async def generate_sow(self, sow_id: UUID = Path(...)): | |
| async with self.__sow_service() as service: | |
| try: | |
| sow = await service.generate_sow(sow_id=sow_id) | |
| return ResponseSOWSingle(status="success", data=SOW(**sow)) | |
| except HTTPException as e: | |
| logger.warning(e) | |
| raise e | |
| except Exception as e: | |
| logger.exception(e) | |
| raise HTTPException(status_code=500, detail="Failed to generate SOW.") | |
| async def download_sow(self, sow_id: UUID = Path(...)): | |
| async with self.__sow_service() as service: | |
| try: | |
| file = await service.download_sow(sow_id=sow_id) | |
| return Response( | |
| content=file, | |
| media_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document", | |
| ) | |
| except HTTPException as e: | |
| logger.warning(e) | |
| raise e | |
| except Exception as e: | |
| logger.exception(e) | |
| raise HTTPException(status_code=500, detail="Failed to download SOW.") | |
| async def upload_sow( | |
| self, | |
| files: List[UploadFile] = File(...), | |
| ): | |
| async with self.__sow_service() as service: | |
| try: | |
| data = await service.upload_sow(files=files) | |
| return ResponseSOWSingle(status="success", data=SOW(**data)) | |
| except HTTPException as e: | |
| logger.warning(e) | |
| raise e | |
| except Exception as e: | |
| logger.exception(e) | |
| raise HTTPException(status_code=500, detail="Failed to upload SOW.") | |
| async def upload_sow_by_id( | |
| self, | |
| sow_id: UUID = Path(...), | |
| files: List[UploadFile] = File(...), | |
| ): | |
| async with self.__sow_service() as service: | |
| try: | |
| data = await service.upload_sow(files=files, id=sow_id) | |
| return ResponseSOWSingle(status="success", data=SOW(**data)) | |
| except HTTPException as e: | |
| logger.warning(e) | |
| raise e | |
| except Exception as e: | |
| logger.exception(e) | |
| raise HTTPException(status_code=500, detail="Failed to upload SOW.") | |