Exception Handling

class HTTPException(Exception):
    def __init__(self, status_code: int, detail: str | None = None) -> None:
        """
        HTTP exception with status code and detail message.
        
        Args:
            status_code: HTTP status code (e.g., 404, 500)
            detail: Optional error detail message (defaults to standard HTTP phrase)
        """
    
    def __str__(self) -> str:
        """String representation of the exception."""
    
    def __repr__(self) -> str:
        """Detailed string representation for debugging."""

class WebSocketException(Exception):
    def __init__(self, code: int, reason: str | None = None) -> None:
        """
        WebSocket exception with close code and reason.
        
        Args:
            code: WebSocket close code (e.g., 1000, 1001, 1002)
            reason: Optional close reason message
        """
    
    def __str__(self) -> str:
        """String representation of the exception."""
    
    def __repr__(self) -> str:
        """Detailed string representation for debugging."""

from robyn import Robyn, status_codes
from robyn.exceptions import HTTPException

app = Robyn(__file__)

@app.get("/user/<user_id>")
def get_user(request):
    user_id = request.path_params["user_id"]
    
    # Validate user ID
    if not user_id.isdigit():
        raise HTTPException(
            status_code=status_codes.HTTP_400_BAD_REQUEST,
            detail="User ID must be a number"
        )
    
    user_id = int(user_id)
    
    # Check if user exists (example)
    if user_id < 1 or user_id > 1000:
        raise HTTPException(
            status_code=status_codes.HTTP_404_NOT_FOUND,
            detail=f"User {user_id} not found"
        )
    
    # Simulate authorization check
    auth_header = request.headers.get("Authorization")
    if not auth_header:
        raise HTTPException(
            status_code=status_codes.HTTP_401_UNAUTHORIZED,
            detail="Authentication required"
        )
    
    return {"user_id": user_id, "name": f"User {user_id}"}

@app.get("/admin/users")
def admin_users(request):
    # Check admin permissions
    user_role = request.headers.get("X-User-Role")
    if user_role != "admin":
        raise HTTPException(
            status_code=status_codes.HTTP_403_FORBIDDEN,
            detail="Admin access required"
        )
    
    return {"users": ["alice", "bob", "charlie"]}

app.start()

from robyn import Robyn, Response, status_codes
from robyn.exceptions import HTTPException

app = Robyn(__file__)

@app.exception
def handle_exception(request, exception):
    """Global exception handler for all unhandled exceptions."""
    if isinstance(exception, HTTPException):
        return Response(
            status_code=exception.status_code,
            headers={"Content-Type": "application/json"},
            description=f'{{"error": "{exception.detail}"}}'
        )
    else:
        # Handle unexpected exceptions
        return Response(
            status_code=status_codes.HTTP_500_INTERNAL_SERVER_ERROR,
            headers={"Content-Type": "application/json"},
            description='{"error": "Internal server error"}'
        )

@app.get("/error-demo/<error_type>")
def error_demo(request):
    error_type = request.path_params["error_type"]
    
    if error_type == "not_found":
        raise HTTPException(404, "Resource not found")
    elif error_type == "bad_request":
        raise HTTPException(400, "Invalid request parameters")
    elif error_type == "unauthorized":
        raise HTTPException(401)  # Uses default message
    elif error_type == "server_error":
        # This will trigger the generic exception handler
        raise ValueError("Something went wrong")
    
    return {"message": "No error triggered"}

app.start()

from robyn import Robyn
from robyn.ws import WebSocket
from robyn.exceptions import WebSocketException

app = Robyn(__file__)
websocket = WebSocket(app, "/ws")

@websocket.on("connect")
def on_connect(websocket_connector):
    # Validate connection (example)
    auth_token = websocket_connector.query_params.get("token")
    if not auth_token:
        raise WebSocketException(
            code=1008,  # Policy violation
            reason="Authentication token required"
        )
    
    print(f"Client {websocket_connector.id} connected")

@websocket.on("message")
def on_message(websocket_connector, message):
    try:
        # Process message
        if message == "close":
            raise WebSocketException(
                code=1000,  # Normal closure
                reason="Client requested close"
            )
        
        # Echo message back
        websocket_connector.sync_send_to(websocket_connector.id, f"Echo: {message}")
        
    except Exception as e:
        # Handle processing errors
        raise WebSocketException(
            code=1011,  # Unexpected condition
            reason=f"Message processing error: {str(e)}"
        )

@websocket.on("close")
def on_close(websocket_connector):
    print(f"Client {websocket_connector.id} disconnected")

app.add_web_socket("/ws", websocket)
app.start()

from robyn import Robyn, status_codes
from robyn.exceptions import HTTPException

# Custom business logic exceptions
class UserNotFoundError(HTTPException):
    def __init__(self, user_id: int):
        super().__init__(
            status_code=status_codes.HTTP_404_NOT_FOUND,
            detail=f"User with ID {user_id} not found"
        )

class InsufficientPermissionsError(HTTPException):
    def __init__(self, required_role: str):
        super().__init__(
            status_code=status_codes.HTTP_403_FORBIDDEN,
            detail=f"Requires {required_role} role"
        )

class ValidationError(HTTPException):
    def __init__(self, field: str, message: str):
        super().__init__(
            status_code=status_codes.HTTP_422_UNPROCESSABLE_ENTITY,
            detail=f"Validation error in {field}: {message}"
        )

app = Robyn(__file__)

@app.post("/users")
def create_user(request):
    user_data = request.json()
    
    # Validation
    if not user_data.get("email"):
        raise ValidationError("email", "Email is required")
    
    if "@" not in user_data["email"]:
        raise ValidationError("email", "Invalid email format")
    
    # Create user logic here
    return {"message": "User created", "id": 123}

@app.get("/user/<user_id>/profile")
def get_user_profile(request):
    user_id = int(request.path_params["user_id"])
    
    # Check if user exists
    if user_id > 1000:  # Example check
        raise UserNotFoundError(user_id)
    
    # Check permissions
    user_role = request.headers.get("X-User-Role")
    if user_role not in ["admin", "user"]:
        raise InsufficientPermissionsError("user")
    
    return {"user_id": user_id, "profile": "User profile data"}

app.start()

import logging
from robyn import Robyn, Response, status_codes
from robyn.exceptions import HTTPException, WebSocketException

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

app = Robyn(__file__)

@app.exception
def log_and_handle_exception(request, exception):
    """Exception handler with logging for monitoring."""
    
    if isinstance(exception, HTTPException):
        # Log HTTP exceptions
        logger.warning(
            f"HTTP Exception: {exception.status_code} - {exception.detail} "
            f"[{request.method} {request.url.path}]"
        )
        
        return Response(
            status_code=exception.status_code,
            headers={"Content-Type": "application/json"},
            description=f'{{"error": "{exception.detail}"}}'
        )
    
    else:
        # Log unexpected exceptions
        logger.error(
            f"Unhandled Exception: {type(exception).__name__} - {str(exception)} "
            f"[{request.method} {request.url.path}]",
            exc_info=True
        )
        
        return Response(
            status_code=status_codes.HTTP_500_INTERNAL_SERVER_ERROR,
            headers={"Content-Type": "application/json"},
            description='{"error": "Internal server error"}'
        )

@app.get("/test/<scenario>")
def test_exceptions(request):
    scenario = request.path_params["scenario"]
    
    scenarios = {
        "http_400": lambda: HTTPException(400, "Bad request test"),
        "http_404": lambda: HTTPException(404, "Not found test"),
        "http_500": lambda: HTTPException(500, "Server error test"),
        "runtime": lambda: RuntimeError("Runtime error test"),
        "value": lambda: ValueError("Value error test"),
    }
    
    if scenario in scenarios:
        raise scenarios[scenario]()
    else:
        return {"message": f"Unknown scenario: {scenario}"}

app.start()