Spaces:
Sleeping
Sleeping
File size: 6,339 Bytes
a83c934 |
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 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 |
"""Custom error handlers for user-friendly error messages.
Provides consistent, helpful error messages for common failure scenarios.
"""
from typing import Dict, Any, Optional
from fastapi import Request, status
from fastapi.responses import JSONResponse
from openai import APIStatusError, APITimeoutError, APIConnectionError
from qdrant_client.http.exceptions import UnexpectedResponse
from src.utils.logger import get_logger
logger = get_logger(__name__)
class ErrorMessages:
"""Centralized error message templates for user-friendly responses"""
# Authentication errors
SESSION_EXPIRED = "Your session has expired. Please log in again."
INVALID_CREDENTIALS = "Invalid email or password. Please try again."
UNAUTHORIZED = "You must be logged in to access this resource."
ACCOUNT_EXISTS = "An account with this email already exists."
# API errors
API_TIMEOUT = "The service is taking longer than expected. Please try again."
API_UNAVAILABLE = "The AI service is temporarily unavailable. Please try again in a few moments."
API_RATE_LIMIT = "Too many requests. Please wait a moment before trying again."
# Vector database errors
VECTOR_DB_ERROR = "Unable to search book content. Please try again."
NO_RESULTS = "No relevant content found for your question. Try rephrasing or asking something else."
# Validation errors
INVALID_INPUT = "Invalid input provided. Please check your data and try again."
MESSAGE_TOO_LONG = "Your message is too long. Please keep it under 10,000 characters."
SELECTED_TEXT_REQUIRED = "Please select some text first before asking a question about it."
# Database errors
DATABASE_ERROR = "A database error occurred. Please try again."
CONNECTION_ERROR = "Unable to connect to the database. Please try again later."
# Generic errors
INTERNAL_ERROR = "An unexpected error occurred. Our team has been notified."
NOT_FOUND = "The requested resource was not found."
def create_error_response(
message: str,
status_code: int = status.HTTP_500_INTERNAL_SERVER_ERROR,
details: Optional[Dict[str, Any]] = None
) -> JSONResponse:
"""Create a standardized error response.
Args:
message: User-friendly error message
status_code: HTTP status code
details: Optional additional error details (for debugging)
Returns:
JSONResponse with error information
"""
content = {
"error": True,
"message": message,
"status_code": status_code
}
if details:
content["details"] = details
return JSONResponse(
status_code=status_code,
content=content
)
async def openai_error_handler(request: Request, exc: Exception) -> JSONResponse:
"""Handle OpenAI API errors with user-friendly messages.
Args:
request: The request that caused the error
exc: The exception that was raised
Returns:
JSONResponse with user-friendly error
"""
if isinstance(exc, APIStatusError):
if exc.status_code == 429:
logger.warning(f"OpenAI rate limit hit for {request.url.path}")
return create_error_response(
ErrorMessages.API_RATE_LIMIT,
status.HTTP_429_TOO_MANY_REQUESTS,
{"retry_after": exc.response.headers.get("Retry-After", "60")}
)
elif exc.status_code >= 500:
logger.error(f"OpenAI server error: {exc.message}")
return create_error_response(
ErrorMessages.API_UNAVAILABLE,
status.HTTP_503_SERVICE_UNAVAILABLE
)
else:
logger.error(f"OpenAI API error ({exc.status_code}): {exc.message}")
return create_error_response(
ErrorMessages.API_UNAVAILABLE,
status.HTTP_502_BAD_GATEWAY
)
elif isinstance(exc, APITimeoutError):
logger.warning(f"OpenAI API timeout for {request.url.path}")
return create_error_response(
ErrorMessages.API_TIMEOUT,
status.HTTP_504_GATEWAY_TIMEOUT
)
elif isinstance(exc, APIConnectionError):
logger.error(f"OpenAI connection error: {exc}")
return create_error_response(
ErrorMessages.API_UNAVAILABLE,
status.HTTP_503_SERVICE_UNAVAILABLE
)
# Default OpenAI error
logger.error(f"Unexpected OpenAI error: {exc}")
return create_error_response(
ErrorMessages.INTERNAL_ERROR,
status.HTTP_500_INTERNAL_SERVER_ERROR
)
async def qdrant_error_handler(request: Request, exc: Exception) -> JSONResponse:
"""Handle Qdrant errors with user-friendly messages.
Args:
request: The request that caused the error
exc: The exception that was raised
Returns:
JSONResponse with user-friendly error
"""
if isinstance(exc, UnexpectedResponse):
logger.error(f"Qdrant error for {request.url.path}: {exc}")
return create_error_response(
ErrorMessages.VECTOR_DB_ERROR,
status.HTTP_503_SERVICE_UNAVAILABLE
)
logger.error(f"Unexpected Qdrant error: {exc}")
return create_error_response(
ErrorMessages.VECTOR_DB_ERROR,
status.HTTP_500_INTERNAL_SERVER_ERROR
)
async def validation_error_handler(request: Request, exc: Exception) -> JSONResponse:
"""Handle validation errors with user-friendly messages.
Args:
request: The request that caused the error
exc: The exception that was raised
Returns:
JSONResponse with user-friendly error
"""
from pydantic import ValidationError
if isinstance(exc, ValidationError):
# Extract field-specific errors
errors = []
for error in exc.errors():
field = " -> ".join(str(loc) for loc in error["loc"])
message = error["msg"]
errors.append(f"{field}: {message}")
logger.warning(f"Validation error for {request.url.path}: {errors}")
return create_error_response(
ErrorMessages.INVALID_INPUT,
status.HTTP_422_UNPROCESSABLE_ENTITY,
{"validation_errors": errors}
)
return create_error_response(
ErrorMessages.INVALID_INPUT,
status.HTTP_400_BAD_REQUEST
)
|