tessl/pypi-litestar

Litestar is a powerful, flexible yet opinionated ASGI web framework specifically focused on building high-performance APIs.

—

Pending

Overview

Eval results

Files

Exception Handling

Name: tessl/pypi-litestar
Author: tessl

Framework-specific exceptions and error handling including HTTP exceptions, validation errors, and custom exception handlers. Litestar provides structured error responses and comprehensive exception handling with debugging support.

Capabilities

Base Exceptions

Core exception classes that serve as the foundation for all framework exceptions.

class LitestarException(Exception):
    """Base exception for all Litestar-specific errors."""
    
    def __init__(self, detail: str = "", *args: object) -> None:
        """
        Initialize base Litestar exception.

        Parameters:
        - detail: Human-readable error description
        - *args: Additional exception arguments
        """
        super().__init__(detail, *args)
        self.detail = detail

class LitestarWarning(UserWarning):
    """Base warning class for Litestar framework warnings."""

class MissingDependencyException(LitestarException, ImportError):
    """Exception raised when a required dependency is not available."""

class SerializationException(LitestarException):
    """Exception raised during data serialization/deserialization."""

class ImproperlyConfiguredException(LitestarException):
    """Exception raised when the application is improperly configured."""

HTTP Exceptions

HTTP-specific exceptions that generate appropriate HTTP error responses.

class HTTPException(LitestarException):
    def __init__(
        self,
        detail: str = "",
        *,
        status_code: int = 500,
        headers: dict[str, str] | None = None,
        extra: dict[str, Any] | list[Any] | None = None,
    ):
        """
        Base HTTP exception.

        Parameters:
        - detail: Human-readable error description
        - status_code: HTTP status code
        - headers: Additional response headers
        - extra: Additional data to include in error response
        """
        super().__init__(detail)
        self.status_code = status_code
        self.headers = headers or {}
        self.extra = extra

    def to_response(self) -> Response:
        """Convert exception to HTTP response."""

class ClientException(HTTPException):
    """Base class for 4xx client error exceptions."""
    status_code = 400

class InternalServerException(HTTPException):
    """Base class for 5xx server error exceptions."""
    status_code = 500

# 4xx Client Error Exceptions
class BadRequestException(ClientException):
    """400 Bad Request exception."""
    status_code = HTTP_400_BAD_REQUEST

class NotAuthorizedException(ClientException):
    """401 Unauthorized exception."""
    status_code = HTTP_401_UNAUTHORIZED

class PermissionDeniedException(ClientException):
    """403 Forbidden exception."""
    status_code = HTTP_403_FORBIDDEN

class NotFoundException(ClientException):
    """404 Not Found exception."""
    status_code = HTTP_404_NOT_FOUND

class MethodNotAllowedException(ClientException):
    """405 Method Not Allowed exception."""
    status_code = HTTP_405_METHOD_NOT_ALLOWED
    
    def __init__(
        self,
        detail: str = "",
        *,
        method: str,
        path: str,
        **kwargs: Any,
    ):
        """
        Method not allowed exception.

        Parameters:
        - detail: Error description
        - method: HTTP method that was not allowed
        - path: Request path
        """
        super().__init__(detail, **kwargs)
        self.method = method
        self.path = path

class NotAcceptableException(ClientException):
    """406 Not Acceptable exception."""
    status_code = HTTP_406_NOT_ACCEPTABLE

class RequestTimeoutException(ClientException):
    """408 Request Timeout exception."""
    status_code = HTTP_408_REQUEST_TIMEOUT

class ConflictException(ClientException):
    """409 Conflict exception."""
    status_code = HTTP_409_CONFLICT

class ValidationException(ClientException):
    """422 Unprocessable Entity exception for validation errors."""
    status_code = HTTP_422_UNPROCESSABLE_ENTITY

class TooManyRequestsException(ClientException):
    """429 Too Many Requests exception."""
    status_code = HTTP_429_TOO_MANY_REQUESTS

# 5xx Server Error Exceptions  
class InternalServerException(InternalServerException):
    """500 Internal Server Error exception."""
    status_code = HTTP_500_INTERNAL_SERVER_ERROR

class NotImplementedException(InternalServerException):
    """501 Not Implemented exception."""
    status_code = HTTP_501_NOT_IMPLEMENTED

class BadGatewayException(InternalServerException):
    """502 Bad Gateway exception."""
    status_code = HTTP_502_BAD_GATEWAY

class ServiceUnavailableException(InternalServerException):
    """503 Service Unavailable exception."""
    status_code = HTTP_503_SERVICE_UNAVAILABLE

WebSocket Exceptions

Exceptions specific to WebSocket connections.

class WebSocketException(LitestarException):
    def __init__(self, detail: str = "", code: int = 1011):
        """
        WebSocket exception.

        Parameters:
        - detail: Error description
        - code: WebSocket close code
        """
        super().__init__(detail)
        self.code = code

class WebSocketDisconnect(WebSocketException):
    """Exception raised when WebSocket connection is unexpectedly closed."""
    
    def __init__(self, code: int = 1000, reason: str | None = None):
        """
        WebSocket disconnect exception.

        Parameters:
        - code: WebSocket close code
        - reason: Disconnect reason
        """
        detail = f"WebSocket disconnected (code: {code})"
        if reason:
            detail += f" - {reason}"
        super().__init__(detail, code)
        self.reason = reason

DTO Exceptions

Exceptions related to Data Transfer Object operations.

class DTOFactoryException(LitestarException):
    """Exception raised during DTO factory operations."""

class InvalidAnnotationException(DTOFactoryException):
    """Exception raised when encountering invalid type annotations."""

Route and Template Exceptions

Exceptions related to routing and template processing.

class NoRouteMatchFoundException(InternalServerException):
    """Exception raised when no route matches the request."""
    status_code = HTTP_404_NOT_FOUND

class TemplateNotFoundException(InternalServerException):
    """Exception raised when a template cannot be found."""
    status_code = HTTP_500_INTERNAL_SERVER_ERROR

Exception Handlers

Functions and classes for handling exceptions and generating error responses.

def create_exception_response(
    request: Request,
    exc: Exception,
) -> Response:
    """
    Create an HTTP response from an exception.

    Parameters:
    - request: HTTP request that caused the exception
    - exc: Exception to convert to response

    Returns:
    HTTP response representing the exception
    """

class ExceptionHandler:
    def __init__(
        self,
        handler: Callable[[Request, Exception], Response | Awaitable[Response]],
        *,
        status_code: int | None = None,
        media_type: MediaType | str | None = None,
    ):
        """
        Exception handler configuration.

        Parameters:
        - handler: Function to handle the exception
        - status_code: Override status code
        - media_type: Response media type
        """

def exception_handler(
    exc_class: type[Exception],
    *,
    status_code: int | None = None,
    media_type: MediaType | str | None = None,
) -> Callable[[T], T]:
    """
    Decorator to register exception handlers.

    Parameters:
    - exc_class: Exception class to handle
    - status_code: Override status code
    - media_type: Response media type

    Returns:
    Decorator function
    """

Validation Error Details

Structured validation error information for detailed error responses.

class ValidationErrorDetail:
    def __init__(
        self,
        message: str,
        key: str,
        source: Literal["body", "query", "path", "header", "cookie"] | None = None,
    ):
        """
        Validation error detail.

        Parameters:
        - message: Error message
        - key: Field or parameter name that failed validation
        - source: Source of the invalid data
        """
        self.message = message
        self.key = key
        self.source = source

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary representation."""
        return {
            "message": self.message,
            "key": self.key,
            "source": self.source,
        }

Usage Examples

Basic Exception Handling

from litestar import get
from litestar.exceptions import HTTPException, NotFoundException
from litestar.status_codes import HTTP_400_BAD_REQUEST

@get("/users/{user_id:int}")
def get_user(user_id: int) -> dict:
    if user_id < 1:
        raise HTTPException(
            detail="User ID must be positive",
            status_code=HTTP_400_BAD_REQUEST,
            extra={"provided_id": user_id}
        )
    
    # Simulate user lookup
    if user_id == 999:
        raise NotFoundException("User not found")
    
    return {"id": user_id, "name": "Alice"}

@get("/protected")
def protected_resource() -> dict:
    # Check authentication (simplified)
    authenticated = False
    if not authenticated:
        raise NotAuthorizedException("Authentication required")
    
    return {"data": "secret information"}

Custom Exception Classes

from litestar.exceptions import HTTPException
from litestar.status_codes import HTTP_402_PAYMENT_REQUIRED

class InsufficientCreditsException(HTTPException):
    """Custom exception for payment/credits system."""
    status_code = HTTP_402_PAYMENT_REQUIRED
    
    def __init__(self, required_credits: int, available_credits: int):
        detail = f"Insufficient credits: need {required_credits}, have {available_credits}"
        super().__init__(
            detail=detail,
            extra={
                "required_credits": required_credits,
                "available_credits": available_credits,
                "credit_purchase_url": "/purchase-credits"
            }
        )

class RateLimitExceededException(HTTPException):
    """Custom exception for rate limiting."""
    status_code = HTTP_429_TOO_MANY_REQUESTS
    
    def __init__(self, limit: int, window: int, retry_after: int):
        detail = f"Rate limit exceeded: {limit} requests per {window} seconds"
        super().__init__(
            detail=detail,
            headers={"Retry-After": str(retry_after)},
            extra={
                "limit": limit,
                "window": window,
                "retry_after": retry_after
            }
        )

@get("/premium-feature")
def premium_feature(user_credits: int = 10) -> dict:
    required_credits = 50
    
    if user_credits < required_credits:
        raise InsufficientCreditsException(required_credits, user_credits)
    
    return {"result": "premium feature accessed"}

Global Exception Handlers

from litestar import Litestar
from litestar.exceptions import HTTPException, ValidationException
from litestar.response import Response
import logging

logger = logging.getLogger(__name__)

def generic_exception_handler(request: Request, exc: Exception) -> Response:
    """Handle unexpected exceptions."""
    logger.exception("Unhandled exception occurred")
    
    return Response(
        content={
            "error": "Internal server error",
            "detail": "An unexpected error occurred",
            "request_id": getattr(request.state, "request_id", None)
        },
        status_code=500,
        headers={"X-Error-Type": "UnhandledException"}
    )

def validation_exception_handler(request: Request, exc: ValidationException) -> Response:
    """Handle validation errors with detailed information."""
    return Response(
        content={
            "error": "Validation failed",
            "detail": exc.detail,
            "validation_errors": exc.extra or [],
            "path": str(request.url.path),
            "method": request.method
        },
        status_code=exc.status_code,
        headers={"X-Error-Type": "ValidationError"}
    )

def http_exception_handler(request: Request, exc: HTTPException) -> Response:
    """Handle HTTP exceptions with consistent format."""
    return Response(
        content={
            "error": exc.__class__.__name__,
            "detail": exc.detail,
            "status_code": exc.status_code,
            **({"extra": exc.extra} if exc.extra else {})
        },
        status_code=exc.status_code,
        headers=exc.headers
    )

app = Litestar(
    route_handlers=[...],
    exception_handlers={
        HTTPException: http_exception_handler,
        ValidationException: validation_exception_handler,
        Exception: generic_exception_handler,
    }
)

Validation Exception with Details

from litestar.exceptions import ValidationException, ValidationErrorDetail

@post("/users")
def create_user(data: dict) -> dict:
    errors = []
    
    # Validate required fields
    if not data.get("name"):
        errors.append(ValidationErrorDetail(
            message="Name is required",
            key="name",
            source="body"
        ))
    
    if not data.get("email"):
        errors.append(ValidationErrorDetail(
            message="Email is required", 
            key="email",
            source="body"
        ))
    elif "@" not in data["email"]:
        errors.append(ValidationErrorDetail(
            message="Invalid email format",
            key="email", 
            source="body"
        ))
    
    if errors:
        raise ValidationException(
            detail="User validation failed",
            extra=[error.to_dict() for error in errors]
        )
    
    return {"id": 123, "name": data["name"], "email": data["email"]}

WebSocket Exception Handling

from litestar import websocket, WebSocket
from litestar.exceptions import WebSocketException, WebSocketDisconnect

@websocket("/ws")
async def websocket_handler(websocket: WebSocket) -> None:
    try:
        await websocket.accept()
        
        while True:
            message = await websocket.receive_json()
            
            # Validate message format
            if not isinstance(message, dict) or "type" not in message:
                raise WebSocketException(
                    "Invalid message format: must be JSON object with 'type' field",
                    code=1003  # Unsupported data
                )
            
            # Process message
            await websocket.send_json({"status": "received", "data": message})
            
    except WebSocketDisconnect as exc:
        logger.info(f"WebSocket disconnected: {exc.reason}")
    except WebSocketException as exc:
        logger.error(f"WebSocket error: {exc.detail}")
        await websocket.close(code=exc.code, reason=exc.detail)
    except Exception as exc:
        logger.exception("Unexpected error in WebSocket handler")
        await websocket.close(code=1011, reason="Internal error")

Exception Handler Decorator

from litestar.exceptions import exception_handler

@exception_handler(ValueError)
def handle_value_error(request: Request, exc: ValueError) -> Response:
    """Handle ValueError exceptions."""
    return Response(
        content={
            "error": "Invalid value",
            "detail": str(exc),
            "path": str(request.url.path)
        },
        status_code=400
    )

@exception_handler(KeyError)
def handle_key_error(request: Request, exc: KeyError) -> Response:
    """Handle KeyError exceptions."""
    return Response(
        content={
            "error": "Missing required field",
            "detail": f"Required field not found: {exc}",
            "path": str(request.url.path)
        },
        status_code=400
    )

# Apply to specific routes
@get("/data/{key:str}", exception_handlers={KeyError: handle_key_error})
def get_data(key: str) -> dict:
    data = {"user": "alice", "admin": "bob"}
    return {"value": data[key]}  # May raise KeyError

Error Response Formatting

from litestar.response import Response
from litestar.status_codes import *

class APIErrorResponse:
    """Standardized API error response format."""
    
    def __init__(
        self,
        error_code: str,
        message: str,
        details: dict[str, Any] | None = None,
        status_code: int = 500,
    ):
        self.error_code = error_code
        self.message = message
        self.details = details or {}
        self.status_code = status_code
    
    def to_response(self) -> Response:
        content = {
            "success": False,
            "error": {
                "code": self.error_code,
                "message": self.message,
                "timestamp": datetime.utcnow().isoformat(),
                **self.details
            }
        }
        return Response(content=content, status_code=self.status_code)

def api_exception_handler(request: Request, exc: HTTPException) -> Response:
    """Convert HTTP exceptions to standardized API error format."""
    error_mapping = {
        400: "BAD_REQUEST",
        401: "UNAUTHORIZED", 
        403: "FORBIDDEN",
        404: "NOT_FOUND",
        422: "VALIDATION_ERROR",
        429: "RATE_LIMIT_EXCEEDED",
        500: "INTERNAL_ERROR",
    }
    
    error_code = error_mapping.get(exc.status_code, "UNKNOWN_ERROR")
    
    api_error = APIErrorResponse(
        error_code=error_code,
        message=exc.detail,
        details=exc.extra or {},
        status_code=exc.status_code
    )
    
    return api_error.to_response()

Debugging and Development Exception Handling

import traceback
from litestar import Litestar

def debug_exception_handler(request: Request, exc: Exception) -> Response:
    """Detailed exception handler for development."""
    return Response(
        content={
            "error": exc.__class__.__name__,
            "message": str(exc),
            "traceback": traceback.format_exc().split('\n'),
            "request": {
                "method": request.method,
                "url": str(request.url),
                "headers": dict(request.headers),
                "path_params": request.path_params,
                "query_params": dict(request.query_params),
            }
        },
        status_code=500,
        headers={"Content-Type": "application/json"}
    )

def create_app(debug: bool = False) -> Litestar:
    exception_handlers = {}
    
    if debug:
        exception_handlers[Exception] = debug_exception_handler
    else:
        exception_handlers[Exception] = generic_exception_handler
    
    return Litestar(
        route_handlers=[...],
        exception_handlers=exception_handlers,
        debug=debug
    )

Types

# Exception handler types
ExceptionHandler = Callable[[Request, Exception], Response | Awaitable[Response]]
ExceptionHandlersMap = dict[type[Exception], ExceptionHandler]

# Validation error types
ValidationSource = Literal["body", "query", "path", "header", "cookie"]

# WebSocket close codes (RFC 6455)
WS_CLOSE_CODES = {
    1000: "Normal Closure",
    1001: "Going Away", 
    1002: "Protocol Error",
    1003: "Unsupported Data",
    1007: "Invalid Frame Payload Data",
    1008: "Policy Violation",
    1009: "Message Too Big",
    1010: "Mandatory Extension",
    1011: "Internal Error",
    1015: "TLS Handshake",
}

# Custom application close codes (4000-4999)
WS_CUSTOM_CLOSE_CODES = {
    4001: "Unauthorized",
    4003: "Forbidden", 
    4004: "Not Found",
    4008: "Rate Limited",
    4009: "Invalid Request",
}

Install with Tessl CLI