from datetime import datetime from typing import Optional from fastapi import HTTPException, Request from fastapi.exceptions import RequestValidationError from fastapi.responses import JSONResponse class APIException(Exception): """Base API Exception""" def __init__( self, status_code: int, message: str, error_type: Optional[str] = None, details: Optional[dict] = None, ): self.status_code = status_code self.message = message self.error_type = error_type or self.__class__.__name__ self.details = details or {} class UnauthorizedException(APIException): """Exception for unauthorized access (401)""" def __init__( self, message: str = "Authentication required", details: Optional[dict] = None, ): super().__init__( status_code=401, message=message, error_type="UnauthorizedException", details=details, ) async def api_exception_handler(request: Request, exc: APIException) -> JSONResponse: """Custom exception handler for APIException""" return JSONResponse( status_code=exc.status_code, content={ "status_code": exc.status_code, "success": False, "timestamp": datetime.utcnow().isoformat(), "cause": { "type": exc.error_type, "message": exc.message, "details": exc.details, "path": str(request.url), }, }, ) async def unauthorized_exception_handler( request: Request, exc: UnauthorizedException ) -> JSONResponse: """Handler specifically for 401 Unauthorized errors""" return JSONResponse( status_code=401, content={ "status_code": 401, "success": False, "timestamp": datetime.utcnow().isoformat(), "cause": { "type": "UnauthorizedException", "message": exc.message, "details": exc.details, "path": str(request.url), }, }, headers={"WWW-Authenticate": "Bearer"}, ) async def general_exception_handler(request: Request, exc: Exception) -> JSONResponse: """Handler for all unhandled exceptions""" return JSONResponse( status_code=500, content={ "status_code": 500, "success": False, "timestamp": datetime.utcnow().isoformat(), "cause": { "type": "InternalServerError", "message": "An unexpected error occurred", "details": {"original_error": str(exc)}, "path": str(request.url), }, }, ) async def http_exception_handler(request: Request, exc: HTTPException) -> JSONResponse: """Handler for FastAPI HTTPException""" # Special handling for 401 errors headers = {} if exc.status_code == 401: headers["WWW-Authenticate"] = "Bearer" return JSONResponse( status_code=exc.status_code, content={ "status_code": exc.status_code, "success": False, "timestamp": datetime.utcnow().isoformat(), "cause": { "type": "HTTPException", "message": exc.detail, "details": {}, "path": str(request.url), }, }, headers=headers, ) async def request_validation_exception_handler( request: Request, exc: RequestValidationError ) -> JSONResponse: """Handler untuk validasi request body (FastAPI/Pydantic)""" errors_serializable = [] for err in exc.errors(): err_copy = err.copy() if "ctx" in err_copy: err_copy["ctx"] = {k: str(v) for k, v in err_copy["ctx"].items()} errors_serializable.append(err_copy) return JSONResponse( status_code=422, content={ "status_code": 422, "success": False, "timestamp": datetime.utcnow().isoformat(), "cause": { "type": "ValidationError", "message": "Validation error", "details": {"errors": errors_serializable}, "path": str(request.url), }, }, )