tessl/pypi-werkzeug
The comprehensive WSGI web application library providing essential utilities and components for building Python web applications.
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
exceptions.mddocs/
Exception Handling
Complete HTTP exception hierarchy for proper error handling and response codes with WSGI integration. These exceptions provide a clean way to handle HTTP errors and generate appropriate responses in web applications.
Capabilities
Base HTTP Exception
The foundation class for all HTTP exceptions with WSGI application integration.
class HTTPException(Exception):
def __init__(self, description=None, response=None):
"""
Base class for all HTTP exceptions.
Parameters:
- description: Custom error description
- response: Pre-built Response object to use
"""
# Class attributes
code: int # HTTP status code
description: str # Default error description
# Properties
name: str # Status code name (e.g., "Not Found")
def get_description(self, environ=None, scope=None):
"""
Get the error description for display.
Parameters:
- environ: WSGI environment (optional)
- scope: ASGI scope (optional)
Returns:
Error description string
"""
def get_body(self, environ=None, scope=None):
"""
Get the HTML error page body.
Parameters:
- environ: WSGI environment (optional)
- scope: ASGI scope (optional)
Returns:
HTML error page content
"""
def get_headers(self, environ=None, scope=None):
"""
Get response headers for the exception.
Parameters:
- environ: WSGI environment (optional)
- scope: ASGI scope (optional)
Returns:
List of (name, value) header tuples
"""
def get_response(self, environ=None, scope=None):
"""
Get a Response object for this exception.
Parameters:
- environ: WSGI environment (optional)
- scope: ASGI scope (optional)
Returns:
Response object with appropriate status and content
"""
def __call__(self, environ, start_response):
"""
WSGI application interface - allows exceptions to be called directly.
Parameters:
- environ: WSGI environment
- start_response: WSGI start_response callable
Returns:
WSGI response iterable
"""4xx Client Error Exceptions
Exceptions for client-side errors (400-499 status codes).
class BadRequest(HTTPException):
"""400 Bad Request - The request cannot be fulfilled due to bad syntax."""
code = 400
description = "The browser (or proxy) sent a request that this server could not understand."
class BadRequestKeyError(BadRequest, KeyError):
"""400 Bad Request caused by missing key in MultiDict-like object."""
def __init__(self, arg=None, *args, **kwargs):
"""
Parameters:
- arg: The missing key that caused the error
"""
class SecurityError(BadRequest):
"""400 Bad Request - Security violation detected."""
class Unauthorized(HTTPException):
"""401 Unauthorized - Authentication is required."""
code = 401
description = "The server could not verify that you are authorized to access the URL requested."
def __init__(self, description=None, response=None, www_authenticate=None):
"""
Parameters:
- description: Custom error description
- response: Pre-built response
- www_authenticate: WWW-Authenticate header value
"""
class Forbidden(HTTPException):
"""403 Forbidden - You don't have permission to access the resource."""
code = 403
description = "You don't have the permission to access the requested resource."
class NotFound(HTTPException):
"""404 Not Found - The requested resource could not be found."""
code = 404
description = "The requested URL was not found on the server."
class MethodNotAllowed(HTTPException):
"""405 Method Not Allowed - The method is not allowed for this resource."""
code = 405
description = "The method is not allowed for the requested URL."
def __init__(self, valid_methods=None, description=None):
"""
Parameters:
- valid_methods: List of allowed HTTP methods
- description: Custom error description
"""
class NotAcceptable(HTTPException):
"""406 Not Acceptable - Cannot satisfy Accept headers."""
code = 406
description = "The resource identified by the request is only capable of generating response entities which have content characteristics not acceptable according to the accept headers sent in the request."
class RequestTimeout(HTTPException):
"""408 Request Timeout - The request took too long to process."""
code = 408
description = "The server closed the network connection because the browser didn't finish the request within the specified time."
class Conflict(HTTPException):
"""409 Conflict - Request conflicts with current state of resource."""
code = 409
description = "A conflict happened while processing the request."
class Gone(HTTPException):
"""410 Gone - The resource is no longer available."""
code = 410
description = "The requested URL is no longer available on this server and there is no forwarding address."
class LengthRequired(HTTPException):
"""411 Length Required - Content-Length header is required."""
code = 411
description = "A request with this method requires a valid Content-Length header."
class PreconditionFailed(HTTPException):
"""412 Precondition Failed - Precondition in headers failed."""
code = 412
description = "The precondition on the request for the URL failed positive evaluation."
class RequestEntityTooLarge(HTTPException):
"""413 Payload Too Large - Request entity is too large."""
code = 413
description = "The data value transmitted exceeds the capacity limit."
class RequestURITooLarge(HTTPException):
"""414 URI Too Long - The request URI is too long."""
code = 414
description = "The length of the requested URL exceeds the capacity limit for this server."
class UnsupportedMediaType(HTTPException):
"""415 Unsupported Media Type - Media type not supported."""
code = 415
description = "The server does not support the media type transmitted in the request."
class RequestedRangeNotSatisfiable(HTTPException):
"""416 Range Not Satisfiable - Cannot satisfy Range header."""
code = 416
description = "The server cannot provide the requested range."
def __init__(self, length=None, units="bytes", description=None):
"""
Parameters:
- length: Total content length
- units: Range units (default: "bytes")
- description: Custom error description
"""
class ExpectationFailed(HTTPException):
"""417 Expectation Failed - Cannot meet Expect header requirements."""
code = 417
description = "The server could not meet the requirements of the Expect header."
class ImATeapot(HTTPException):
"""418 I'm a teapot - RFC 2324 April Fools' joke."""
code = 418
description = "This server is a teapot, not a coffee machine."
class MisdirectedRequest(HTTPException):
"""421 Misdirected Request - Request was directed at wrong server."""
code = 421
description = "The request was directed at a server that is not able to produce a response."
class UnprocessableEntity(HTTPException):
"""422 Unprocessable Entity - Request is well-formed but semantically incorrect."""
code = 422
description = "The request was well-formed but was unable to be followed due to semantic errors."
class Locked(HTTPException):
"""423 Locked - Resource is locked."""
code = 423
description = "The resource that is being accessed is locked."
class FailedDependency(HTTPException):
"""424 Failed Dependency - Request failed due to failure of previous request."""
code = 424
description = "The request failed because it depended on another request and that request failed."
class PreconditionRequired(HTTPException):
"""428 Precondition Required - Precondition headers are required."""
code = 428
description = "This request is required to be conditional."
class TooManyRequests(HTTPException):
"""429 Too Many Requests - Rate limit exceeded."""
code = 429
description = "This user has exceeded an allotted request count."
def __init__(self, description=None, response=None, retry_after=None):
"""
Parameters:
- description: Custom error description
- response: Pre-built response
- retry_after: Seconds to wait before retrying
"""
class RequestHeaderFieldsTooLarge(HTTPException):
"""431 Request Header Fields Too Large - Headers are too large."""
code = 431
description = "One or more header fields exceeds the maximum size."
class UnavailableForLegalReasons(HTTPException):
"""451 Unavailable For Legal Reasons - Content blocked for legal reasons."""
code = 451
description = "Unavailable for legal reasons."5xx Server Error Exceptions
Exceptions for server-side errors (500-599 status codes).
class InternalServerError(HTTPException):
"""500 Internal Server Error - Generic server error."""
code = 500
description = "The server encountered an internal error and was unable to complete your request."
class NotImplemented(HTTPException):
"""501 Not Implemented - Functionality not implemented."""
code = 501
description = "The server does not support the action requested by the browser."
class BadGateway(HTTPException):
"""502 Bad Gateway - Invalid response from upstream server."""
code = 502
description = "The proxy server received an invalid response from an upstream server."
class ServiceUnavailable(HTTPException):
"""503 Service Unavailable - Service temporarily unavailable."""
code = 503
description = "The server is temporarily unable to service your request due to maintenance downtime or capacity problems."
def __init__(self, description=None, response=None, retry_after=None):
"""
Parameters:
- description: Custom error description
- response: Pre-built response
- retry_after: Seconds until service may be available again
"""
class GatewayTimeout(HTTPException):
"""504 Gateway Timeout - Timeout from upstream server."""
code = 504
description = "The connection to an upstream server timed out."
class HTTPVersionNotSupported(HTTPException):
"""505 HTTP Version Not Supported - HTTP version not supported."""
code = 505
description = "The server does not support the HTTP protocol version used in the request."Exception Utilities
Helper functions and classes for working with HTTP exceptions.
def abort(status, *args, **kwargs):
"""
Raise an HTTPException for the given status code.
Parameters:
- status: HTTP status code (int) or Response object
- *args: Arguments passed to exception constructor
- **kwargs: Keyword arguments passed to exception constructor
Raises:
Appropriate HTTPException subclass for the status code
Examples:
- abort(404) → raises NotFound
- abort(500, description="Custom error") → raises InternalServerError with custom description
"""
class Aborter:
def __init__(self, mapping=None, extra=None):
"""
Exception raiser with customizable status code mapping.
Parameters:
- mapping: Dict mapping status codes to exception classes
- extra: Additional exception mappings
"""
def __call__(self, status, *args, **kwargs):
"""
Raise exception for status code.
Parameters:
- status: HTTP status code or Response object
- *args: Arguments for exception constructor
- **kwargs: Keyword arguments for exception constructor
"""
# Default aborter instance
default_exceptions = {
400: BadRequest,
401: Unauthorized,
403: Forbidden,
404: NotFound,
405: MethodNotAllowed,
406: NotAcceptable,
408: RequestTimeout,
409: Conflict,
410: Gone,
# ... complete mapping available
}Usage Examples
Basic Exception Handling
from werkzeug.exceptions import HTTPException, NotFound, BadRequest, InternalServerError
from werkzeug.wrappers import Request, Response
def view_function(request):
"""Example view that may raise exceptions."""
# Check for required parameter
if 'id' not in request.args:
raise BadRequest("Missing 'id' parameter")
user_id = request.args.get('id')
try:
user_id = int(user_id)
except ValueError:
raise BadRequest("Invalid user ID format")
# Simulate database lookup
if user_id == 0:
raise BadRequest("User ID cannot be zero")
elif user_id < 0:
raise NotFound("User not found")
elif user_id > 1000:
raise InternalServerError("Database connection failed")
return f"User {user_id} profile"
def application(environ, start_response):
"""WSGI application with exception handling."""
request = Request(environ)
try:
result = view_function(request)
response = Response(result)
except HTTPException as e:
# HTTP exceptions can be used directly as WSGI applications
return e(environ, start_response)
except Exception as e:
# Convert unexpected exceptions to 500 errors
error = InternalServerError("An unexpected error occurred")
return error(environ, start_response)
return response(environ, start_response)Custom Error Pages
from werkzeug.exceptions import HTTPException, NotFound, InternalServerError
from werkzeug.wrappers import Response
def custom_error_handler(exception, environ=None):
"""Create custom error responses."""
if isinstance(exception, NotFound):
# Custom 404 page
html = f"""
<html>
<head><title>Page Not Found</title></head>
<body>
<h1>Oops! Page Not Found</h1>
<p>The page you're looking for doesn't exist.</p>
<p>Error code: {exception.code}</p>
<a href="/">Go back to homepage</a>
</body>
</html>
"""
return Response(html, status=404, mimetype='text/html')
elif isinstance(exception, InternalServerError):
# Custom 500 page (don't leak internal details in production)
html = """
<html>
<head><title>Server Error</title></head>
<body>
<h1>Something went wrong</h1>
<p>We're working to fix this issue. Please try again later.</p>
</body>
</html>
"""
return Response(html, status=500, mimetype='text/html')
else:
# Use default error page for other exceptions
return exception.get_response(environ)
def application_with_custom_errors(environ, start_response):
"""WSGI app with custom error handling."""
request = Request(environ)
try:
# Your application logic here
if request.path == '/':
response = Response('Hello World!')
elif request.path == '/error':
raise InternalServerError("Intentional error for testing")
else:
raise NotFound()
except HTTPException as e:
response = custom_error_handler(e, environ)
return response(environ, start_response)Using abort() Function
from werkzeug.exceptions import abort
from werkzeug.wrappers import Request, Response
def protected_view(request):
"""Example of using abort() for concise error handling."""
# Check authentication
if 'Authorization' not in request.headers:
abort(401, description="Authentication required")
# Check permissions
if not user_has_permission(request):
abort(403, description="Insufficient permissions")
# Validate input
if not request.json:
abort(400, description="JSON data required")
data = request.json
if 'name' not in data:
abort(400, description="Missing 'name' field")
# Process valid request
return {"message": f"Hello {data['name']}!"}
def user_has_permission(request):
"""Mock permission check."""
# In real application, implement proper permission checking
return request.headers.get('Authorization') == 'Bearer valid-token'
def rest_api_app(environ, start_response):
"""REST API with abort-based error handling."""
request = Request(environ)
try:
if request.method == 'POST' and request.path == '/api/users':
result = protected_view(request)
response = Response(json.dumps(result), mimetype='application/json')
else:
abort(404, description="API endpoint not found")
except HTTPException as e:
# Return JSON error responses for API
error_data = {
'error': e.name,
'code': e.code,
'description': e.get_description()
}
response = Response(
json.dumps(error_data),
status=e.code,
mimetype='application/json'
)
return response(environ, start_response)Exception Middleware
import traceback
import logging
from werkzeug.exceptions import HTTPException, InternalServerError
class ExceptionMiddleware:
"""Middleware for comprehensive exception handling."""
def __init__(self, app, debug=False, logger=None):
self.app = app
self.debug = debug
self.logger = logger or logging.getLogger(__name__)
def __call__(self, environ, start_response):
"""WSGI middleware interface."""
try:
return self.app(environ, start_response)
except HTTPException as e:
# Log HTTP exceptions for monitoring
self.logger.warning(f"HTTP {e.code}: {e.description}")
return e(environ, start_response)
except Exception as e:
# Log unexpected exceptions
self.logger.error(f"Unhandled exception: {e}")
self.logger.error(traceback.format_exc())
# Create appropriate error response
if self.debug:
# In debug mode, show full traceback
error_html = f"""
<html>
<head><title>Debug Error</title></head>
<body>
<h1>Unhandled Exception</h1>
<h2>{type(e).__name__}: {e}</h2>
<pre>{traceback.format_exc()}</pre>
</body>
</html>
"""
error = InternalServerError()
error.description = "Debug information available"
response = Response(error_html, status=500, mimetype='text/html')
else:
# In production, use generic error
error = InternalServerError("An unexpected error occurred")
response = error.get_response(environ)
return response(environ, start_response)
# Usage
def create_app(debug=False):
def app(environ, start_response):
request = Request(environ)
if request.path == '/crash':
raise ValueError("Intentional crash for testing")
response = Response(f'Hello from {request.path}')
return response(environ, start_response)
# Wrap with exception middleware
return ExceptionMiddleware(app, debug=debug)Custom Exception Classes
from werkzeug.exceptions import HTTPException, BadRequest
class ValidationError(BadRequest):
"""Custom exception for validation errors."""
def __init__(self, field=None, message=None):
self.field = field
description = f"Validation failed"
if field:
description += f" for field '{field}'"
if message:
description += f": {message}"
super().__init__(description)
class QuotaExceeded(HTTPException):
"""Custom exception for quota/rate limiting."""
code = 429
description = "Request quota exceeded"
def __init__(self, quota_type=None, retry_after=None):
self.quota_type = quota_type
self.retry_after = retry_after
description = "Request quota exceeded"
if quota_type:
description += f" for {quota_type}"
super().__init__(description)
def get_headers(self, environ=None, scope=None):
"""Add Retry-After header."""
headers = super().get_headers(environ, scope)
if self.retry_after:
headers.append(('Retry-After', str(self.retry_after)))
return headers
class APIError(HTTPException):
"""Base class for API-specific errors."""
def get_response(self, environ=None, scope=None):
"""Return JSON error response."""
error_data = {
'error': {
'code': self.code,
'name': self.name,
'description': self.get_description(environ, scope)
}
}
return Response(
json.dumps(error_data, indent=2),
status=self.code,
mimetype='application/json',
headers=self.get_headers(environ, scope)
)
# Usage of custom exceptions
def validate_user_data(data):
"""Validate user data and raise custom exceptions."""
if 'email' not in data:
raise ValidationError('email', 'Email is required')
if '@' not in data['email']:
raise ValidationError('email', 'Invalid email format')
if len(data.get('password', '')) < 8:
raise ValidationError('password', 'Password must be at least 8 characters')
def api_endpoint(request):
"""API endpoint using custom exceptions."""
# Check rate limiting
if exceeded_rate_limit(request):
raise QuotaExceeded('requests', retry_after=60)
# Validate JSON data
if not request.is_json:
raise APIError(description="Content-Type must be application/json")
data = request.get_json()
validate_user_data(data)
# Process valid data
return {'status': 'success', 'user_id': 12345}Exception Testing
from werkzeug.test import Client
from werkzeug.exceptions import NotFound, BadRequest
def test_exceptions():
"""Test exception handling in applications."""
def test_app(environ, start_response):
request = Request(environ)
if request.path == '/404':
raise NotFound("Test 404 error")
elif request.path == '/400':
raise BadRequest("Test 400 error")
else:
response = Response("OK")
return response(environ, start_response)
client = Client(test_app)
# Test normal response
response = client.get('/')
assert response.status_code == 200
assert response.text == "OK"
# Test 404 exception
response = client.get('/404')
assert response.status_code == 404
assert "Test 404 error" in response.text
# Test 400 exception
response = client.get('/400')
assert response.status_code == 400
assert "Test 400 error" in response.text
print("All exception tests passed!")
def test_custom_exceptions():
"""Test custom exception behavior."""
# Test custom exception properties
error = ValidationError('email', 'Invalid format')
assert error.code == 400
assert 'email' in error.description
assert 'Invalid format' in error.description
# Test quota exception with headers
quota_error = QuotaExceeded('API requests', retry_after=120)
headers = quota_error.get_headers()
header_dict = dict(headers)
assert header_dict.get('Retry-After') == '120'
print("Custom exception tests passed!")
if __name__ == '__main__':
test_exceptions()
test_custom_exceptions()Production Error Handling
import os
import sys
import logging
from werkzeug.exceptions import HTTPException, InternalServerError
# Configure logging for production
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s %(levelname)s %(name)s %(message)s',
handlers=[
logging.FileHandler('app.log'),
logging.StreamHandler(sys.stdout)
]
)
def production_error_handler(app):
"""Production-ready error handling wrapper."""
logger = logging.getLogger('error_handler')
is_debug = os.getenv('DEBUG', '').lower() in ('1', 'true', 'yes')
def error_app(environ, start_response):
try:
return app(environ, start_response)
except HTTPException as e:
# Log client errors (4xx) at info level
if 400 <= e.code < 500:
logger.info(f"Client error {e.code}: {e.description}")
# Log server errors (5xx) at error level
else:
logger.error(f"Server error {e.code}: {e.description}")
return e(environ, start_response)
except Exception as e:
# Log all unexpected exceptions as errors
logger.error(f"Unhandled exception: {e}", exc_info=True)
# Create safe error response
if is_debug:
# Show details in debug mode
error = InternalServerError(f"Debug: {type(e).__name__}: {e}")
else:
# Generic message in production
error = InternalServerError("Internal server error")
return error(environ, start_response)
return error_app
# Example production application
def create_production_app():
"""Create production-ready application with proper error handling."""
def base_app(environ, start_response):
request = Request(environ)
# Your application routes here
if request.path == '/':
response = Response('Production app running')
elif request.path == '/health':
response = Response('OK', mimetype='text/plain')
else:
raise NotFound("Endpoint not found")
return response(environ, start_response)
# Wrap with production error handler
return production_error_handler(base_app)
if __name__ == '__main__':
from werkzeug.serving import run_simple
app = create_production_app()
run_simple('localhost', 8000, app)Install with Tessl CLI
npx tessl i tessl/pypi-werkzeug