Exception Handling

class HTTPException(Exception):
    def __init__(
        self,
        status_code: int,
        detail: Any = None,
        headers: Optional[Dict[str, str]] = None
    ) -> None:
        """
        HTTP exception for client and server errors.

        Parameters:
        - status_code: HTTP status code (400-599)
        - detail: Error detail message or structured data
                 Can be string, dict, list, or any JSON-serializable data
        - headers: Additional HTTP headers to include in error response

        Behaviors:
        - Automatically generates JSON error response
        - Includes status code and detail in response body
        - Supports structured error details with validation info
        - Integrates with OpenAPI documentation for error responses
        - Can be caught and handled by custom exception handlers
        """

class WebSocketException(Exception):
    def __init__(self, code: int, reason: Optional[str] = None) -> None:
        """
        WebSocket connection exception.

        Parameters:
        - code: WebSocket close code (1000-4999)
               Standard codes: 1000 (normal), 1001 (going away), 1002 (protocol error), etc.
        - reason: Optional reason string for the close

        Behaviors:
        - Automatically closes WebSocket connection with specified code
        - Sends close frame with code and reason to client
        - Follows WebSocket RFC close code standards
        - Can be caught for custom WebSocket error handling
        """

class RequestValidationError(ValueError):
    """
    Exception raised when request data validation fails.
    
    Contains detailed information about validation errors including:
    - Field names and locations (path, query, header, body)
    - Validation error types and messages
    - Input values that caused errors
    
    Automatically generates 422 Unprocessable Entity responses
    with structured error details.
    """

class ResponseValidationError(ValueError):
    """
    Exception raised when response data validation fails.
    
    Indicates programming errors where endpoint returns data
    that doesn't match the declared response model.
    
    Automatically generates 500 Internal Server Error responses
    in development, helps catch response model mismatches.
    """

class WebSocketRequestValidationError(ValueError):
    """
    Exception raised when WebSocket request validation fails.
    
    Similar to RequestValidationError but for WebSocket connections.
    Automatically closes WebSocket connection with error details.
    """

class FastAPIError(Exception):
    """
    Base exception class for FastAPI framework errors.
    
    Used for framework-level errors and as base class
    for other FastAPI-specific exceptions.
    """

@app.exception_handler(ExceptionClass)
async def custom_exception_handler(request: Request, exc: ExceptionClass) -> Response:
    """
    Custom exception handler function.
    
    Parameters:
    - request: The HTTP request that caused the exception
    - exc: The exception instance that was raised
    
    Returns:
    Custom Response object with appropriate status code and content
    
    Behaviors:
    - Called automatically when specified exception is raised
    - Can return any Response type (JSON, HTML, plain text, etc.)
    - Has access to full request context for logging or custom logic
    - Can modify response headers and status codes
    - Supports both sync and async handler functions
    """

from fastapi import FastAPI, HTTPException, status

app = FastAPI()

@app.get("/items/{item_id}")
async def get_item(item_id: int):
    if item_id < 1:
        raise HTTPException(
            status_code=400,
            detail="Item ID must be positive"
        )
    
    if item_id > 1000:
        raise HTTPException(
            status_code=404,
            detail="Item not found"
        )
    
    # Simulate item not found
    if item_id == 999:
        raise HTTPException(
            status_code=404,
            detail={
                "error": "Item not found",
                "item_id": item_id,
                "suggestion": "Try a different item ID"
            }
        )
    
    return {"item_id": item_id, "name": f"Item {item_id}"}

@app.post("/users/")
async def create_user(user_data: dict):
    # Check for duplicate username
    if user_data.get("username") == "admin":
        raise HTTPException(
            status_code=409,
            detail="Username 'admin' is reserved",
            headers={"X-Error": "username-conflict"}
        )
    
    # Validate user data
    if not user_data.get("email"):
        raise HTTPException(
            status_code=422,
            detail={
                "field": "email",
                "message": "Email is required",
                "code": "missing_field"
            }
        )
    
    return {"message": "User created successfully"}

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse

app = FastAPI()

class UserNotFoundError(Exception):
    def __init__(self, user_id: int):
        self.user_id = user_id
        self.message = f"User {user_id} not found"
        super().__init__(self.message)

class InsufficientPermissionsError(Exception):
    def __init__(self, required_role: str, user_role: str):
        self.required_role = required_role
        self.user_role = user_role
        self.message = f"Required role: {required_role}, user role: {user_role}"
        super().__init__(self.message)

class DatabaseConnectionError(Exception):
    def __init__(self, database: str):
        self.database = database
        self.message = f"Failed to connect to database: {database}"
        super().__init__(self.message)

# Custom exception handlers
@app.exception_handler(UserNotFoundError)
async def user_not_found_handler(request: Request, exc: UserNotFoundError):
    return JSONResponse(
        status_code=404,
        content={
            "error": "user_not_found",
            "message": exc.message,
            "user_id": exc.user_id,
            "timestamp": "2023-01-01T00:00:00Z"
        }
    )

@app.exception_handler(InsufficientPermissionsError)
async def insufficient_permissions_handler(request: Request, exc: InsufficientPermissionsError):
    return JSONResponse(
        status_code=403,
        content={
            "error": "insufficient_permissions",
            "message": exc.message,
            "required_role": exc.required_role,
            "user_role": exc.user_role
        }
    )

@app.exception_handler(DatabaseConnectionError)
async def database_error_handler(request: Request, exc: DatabaseConnectionError):
    # Log the error
    print(f"Database connection failed: {exc.database}")
    
    return JSONResponse(
        status_code=503,
        content={
            "error": "service_unavailable",
            "message": "Database temporarily unavailable",
            "retry_after": 60
        },
        headers={"Retry-After": "60"}
    )

# Using custom exceptions
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # Simulate database connection check
    if not check_database_connection():
        raise DatabaseConnectionError("user_db")
    
    # Simulate user lookup
    user = find_user(user_id)  # Your database query
    if not user:
        raise UserNotFoundError(user_id)
    
    return user

@app.delete("/users/{user_id}")
async def delete_user(user_id: int, current_user: dict):
    # Check permissions
    if current_user.get("role") != "admin":
        raise InsufficientPermissionsError("admin", current_user.get("role", "user"))
    
    # Check if user exists
    if not user_exists(user_id):
        raise UserNotFoundError(user_id)
    
    # Delete user
    delete_user_from_db(user_id)
    return {"message": "User deleted successfully"}

from fastapi import FastAPI, WebSocket, WebSocketDisconnect, WebSocketException
import json

app = FastAPI()

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: str):
    await websocket.accept()
    
    try:
        while True:
            # Receive message
            data = await websocket.receive_text()
            
            try:
                # Parse JSON message
                message = json.loads(data)
            except json.JSONDecodeError:
                # Send error and close connection
                raise WebSocketException(
                    code=1003,  # Unsupported data
                    reason="Invalid JSON format"
                )
            
            # Validate message structure
            if "type" not in message:
                raise WebSocketException(
                    code=1002,  # Protocol error
                    reason="Message must include 'type' field"
                )
            
            # Handle different message types
            if message["type"] == "ping":
                await websocket.send_text(json.dumps({"type": "pong"}))
            
            elif message["type"] == "auth":
                # Validate authentication
                if not validate_token(message.get("token")):
                    raise WebSocketException(
                        code=1008,  # Policy violation
                        reason="Invalid authentication token"
                    )
                await websocket.send_text(json.dumps({"type": "auth_success"}))
            
            elif message["type"] == "data":
                # Process data message
                result = process_data(message.get("payload"))
                await websocket.send_text(json.dumps({
                    "type": "result",
                    "data": result
                }))
            
            else:
                raise WebSocketException(
                    code=1003,  # Unsupported data
                    reason=f"Unknown message type: {message['type']}"
                )
    
    except WebSocketDisconnect:
        print(f"Client {client_id} disconnected normally")
    
    except WebSocketException:
        print(f"Client {client_id} disconnected due to protocol error")
        # Exception automatically closes connection with proper code
    
    except Exception as e:
        print(f"Unexpected error for client {client_id}: {e}")
        # Close connection with internal error code
        await websocket.close(code=1011, reason="Internal server error")

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError, ResponseValidationError
import logging

app = FastAPI()

# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Global exception handler for unhandled exceptions
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    logger.error(f"Unhandled exception: {exc}", exc_info=True)
    
    return JSONResponse(
        status_code=500,
        content={
            "error": "internal_server_error",
            "message": "An unexpected error occurred",
            "request_id": generate_request_id()  # Your ID generation logic
        }
    )

# Custom handler for validation errors
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    # Log validation errors
    logger.warning(f"Validation error on {request.url}: {exc.errors()}")
    
    # Transform validation errors into custom format
    errors = []
    for error in exc.errors():
        errors.append({
            "field": ".".join(str(x) for x in error["loc"]),
            "message": error["msg"],
            "type": error["type"],
            "input": error.get("input")
        })
    
    return JSONResponse(
        status_code=422,
        content={
            "error": "validation_failed",
            "message": "Request validation failed",
            "details": errors,
            "request_id": generate_request_id()
        }
    )

# Handler for response validation errors (development only)
@app.exception_handler(ResponseValidationError)
async def response_validation_exception_handler(request: Request, exc: ResponseValidationError):
    logger.error(f"Response validation error: {exc}", exc_info=True)
    
    return JSONResponse(
        status_code=500,
        content={
            "error": "response_validation_failed",
            "message": "Response data doesn't match declared model",
            "detail": str(exc) if app.debug else "Internal server error"
        }
    )

# Override default HTTP exception handler
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    # Log HTTP exceptions
    logger.info(f"HTTP {exc.status_code} on {request.url}: {exc.detail}")
    
    # Add request ID to all HTTP exceptions
    content = {
        "error": "http_exception",
        "message": exc.detail,
        "status_code": exc.status_code,
        "request_id": generate_request_id()
    }
    
    headers = exc.headers or {}
    headers["X-Request-ID"] = content["request_id"]
    
    return JSONResponse(
        status_code=exc.status_code,
        content=content,
        headers=headers
    )

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse, HTMLResponse
import os

app = FastAPI()

# Environment-specific exception handling
IS_DEVELOPMENT = os.getenv("ENVIRONMENT") == "development"

@app.exception_handler(Exception)
async def environment_aware_handler(request: Request, exc: Exception):
    # Different handling based on environment
    if IS_DEVELOPMENT:
        # Detailed error information in development
        import traceback
        return JSONResponse(
            status_code=500,
            content={
                "error": "internal_server_error",
                "message": str(exc),
                "type": type(exc).__name__,
                "traceback": traceback.format_exc().split("\n")
            }
        )
    else:
        # Minimal error information in production
        logger.error(f"Production error: {exc}", exc_info=True)
        return JSONResponse(
            status_code=500,
            content={
                "error": "internal_server_error",
                "message": "An unexpected error occurred"
            }
        )

# Content-type specific error responses
@app.exception_handler(HTTPException)
async def content_type_aware_handler(request: Request, exc: HTTPException):
    accept_header = request.headers.get("accept", "")
    
    if "text/html" in accept_header:
        # Return HTML error page for browsers
        html_content = f"""
        <html>
            <head><title>Error {exc.status_code}</title></head>
            <body>
                <h1>Error {exc.status_code}</h1>
                <p>{exc.detail}</p>
                <a href="/">Go back to home</a>
            </body>
        </html>
        """
        return HTMLResponse(content=html_content, status_code=exc.status_code)
    
    else:
        # Return JSON for API clients
        return JSONResponse(
            status_code=exc.status_code,
            content={
                "error": f"http_{exc.status_code}",
                "message": exc.detail
            }
        )

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
import logging
import time
import uuid

app = FastAPI()

# Set up structured logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

def generate_request_id():
    return str(uuid.uuid4())

@app.middleware("http")
async def add_request_id(request: Request, call_next):
    request_id = generate_request_id()
    request.state.request_id = request_id
    
    start_time = time.time()
    
    response = await call_next(request)
    
    process_time = time.time() - start_time
    
    # Log request completion
    logger.info(
        f"Request completed",
        extra={
            "request_id": request_id,
            "method": request.method,
            "url": str(request.url),
            "status_code": response.status_code,
            "process_time": process_time
        }
    )
    
    response.headers["X-Request-ID"] = request_id
    return response

@app.exception_handler(HTTPException)
async def logged_http_exception_handler(request: Request, exc: HTTPException):
    request_id = getattr(request.state, "request_id", "unknown")
    
    # Log HTTP exceptions with context
    logger.warning(
        f"HTTP exception occurred",
        extra={
            "request_id": request_id,
            "method": request.method,
            "url": str(request.url),
            "status_code": exc.status_code,
            "detail": exc.detail,
            "user_agent": request.headers.get("user-agent"),
            "client_ip": request.client.host if request.client else None
        }
    )
    
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": f"http_{exc.status_code}",
            "message": exc.detail,
            "request_id": request_id
        },
        headers={"X-Request-ID": request_id}
    )

@app.exception_handler(Exception)
async def logged_exception_handler(request: Request, exc: Exception):
    request_id = getattr(request.state, "request_id", "unknown")
    
    # Log unexpected exceptions with full context
    logger.error(
        f"Unexpected exception occurred",
        extra={
            "request_id": request_id,
            "method": request.method,
            "url": str(request.url),
            "exception_type": type(exc).__name__,
            "exception_message": str(exc),
            "user_agent": request.headers.get("user-agent"),
            "client_ip": request.client.host if request.client else None
        },
        exc_info=True
    )
    
    return JSONResponse(
        status_code=500,
        content={
            "error": "internal_server_error",
            "message": "An unexpected error occurred",
            "request_id": request_id
        },
        headers={"X-Request-ID": request_id}
    )