tessl/pypi-waitress
A production-quality pure-Python WSGI server with robust HTTP protocol support and comprehensive configuration options
—
Pending
error-handling.mddocs/
Error Handling and Utilities
Exception classes, logging utilities, and helper functions for error handling, HTTP date formatting, and server management tasks.
Capabilities
Exception Hierarchy
Comprehensive exception classes for different types of server errors and HTTP status codes.
class Error:
"""
Base HTTP error class for all waitress HTTP errors.
Provides HTTP response generation framework and
integration with HTTP status codes and WSGI response handling.
Attributes:
- code (int): HTTP status code
- reason (str): HTTP reason phrase
"""
def __init__(self, body): ...
def to_response(self, ident=None): ...
def wsgi_response(self, environ, start_response): ...
class BadRequest(Error):
"""
400 Bad Request error.
Raised when client sends malformed or invalid requests
that cannot be processed by the server.
"""
class RequestHeaderFieldsTooLarge(BadRequest):
"""
431 Request Header Fields Too Large error.
Raised when request headers exceed the configured
max_request_header_size limit.
"""
class RequestEntityTooLarge(BadRequest):
"""
413 Request Entity Too Large error.
Raised when request body exceeds the configured
max_request_body_size limit.
"""
class InternalServerError(Error):
"""
500 Internal Server Error.
Raised for unexpected server-side errors during
request processing or WSGI application execution.
"""
class ServerNotImplemented(Error):
"""
501 Not Implemented error.
Raised when the server does not support the functionality
required to fulfill the request (e.g., unsupported HTTP methods).
"""
class TransferEncodingNotImplemented(Exception):
"""
Raised for unsupported Transfer-Encoding values.
Currently only 'chunked' and 'identity' encodings are supported.
This exception is raised when requests use other transfer encodings.
"""HTTP Date Utilities
Functions for parsing and formatting HTTP date headers according to RFC standards.
def build_http_date(when):
"""
Format timestamp as HTTP date string.
Parameters:
- when (float): Unix timestamp or time.time() value
Returns:
str: HTTP date string in RFC 1123 format
Example:
>>> build_http_date(1609459200)
'Fri, 01 Jan 2021 00:00:00 GMT'
"""
def parse_http_date(d):
"""
Parse HTTP date string to Unix timestamp.
Parameters:
- d (str): HTTP date string (RFC 1123, RFC 850, or asctime format)
Returns:
float: Unix timestamp
Raises:
ValueError: For unparseable date strings
Example:
>>> parse_http_date('Fri, 01 Jan 2021 00:00:00 GMT')
1609459200.0
"""Server Management Utilities
Helper functions for server lifecycle management and resource cleanup.
def cleanup_unix_socket(path):
"""
Remove existing Unix domain socket file.
Parameters:
- path (str): Path to Unix socket file
Notes:
- Safe to call on non-existent files
- Only removes socket files, not regular files
- Called automatically during server startup
"""
def find_double_newline(s):
"""
Find HTTP header/body separator in request data.
Parameters:
- s (bytes): HTTP request data
Returns:
int: Position of separator, or -1 if not found
Notes:
- Looks for \r\n\r\n (HTTP standard)
- Also handles \n\n for compatibility
"""
def undquote(value):
"""
Remove HTTP quoted string quotes and handle escaping.
Parameters:
- value (str): Potentially quoted HTTP header value
Returns:
str: Unquoted string with escape sequences processed
Example:
>>> undquote('"hello world"')
'hello world'
>>> undquote('"value with \\"quotes\\""')
'value with "quotes"'
"""Logging Infrastructure
Logger objects for different aspects of server operation with proper categorization.
logger = logging.getLogger('waitress')
"""
Main waitress logger for general server events.
Used for:
- Server startup and shutdown messages
- Configuration information
- General operational events
- Connection management
"""
queue_logger = logging.getLogger('waitress.queue')
"""
Task queue logger for monitoring request processing.
Used for:
- Task queue depth monitoring
- Thread pool status
- Request processing bottlenecks
- Performance diagnostics
"""Platform Compatibility Constants
Constants for cross-platform compatibility and system-specific features.
# Platform detection
WIN: bool
"""True if running on Windows platform, False otherwise."""
# System limits
MAXINT: int
"""Maximum integer size for the current platform."""
# Network protocol support
HAS_IPV6: bool
"""True if IPv6 support is available on this platform."""
IPPROTO_IPV6: int
"""IPv6 protocol constant for socket operations."""
IPV6_V6ONLY: int
"""IPv6-only socket option constant (prevents dual-stack behavior)."""Buffer System Constants
Constants controlling buffer behavior and performance characteristics.
# Buffer copy operations
COPY_BYTES: int = 262144 # 256KB - chunk size for copying data between buffers
# String buffer limits
STRBUF_LIMIT: int = 8192 # 8KB - maximum size for simple string buffersError Handling Examples
Common patterns for handling different types of errors in waitress applications.
Custom Error Handling
from waitress import serve
from waitress.utilities import Error, BadRequest, InternalServerError
import logging
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('myapp')
def error_handling_app(environ, start_response):
"""WSGI app with comprehensive error handling."""
try:
# Your application logic here
path = environ['PATH_INFO']
if path == '/error':
raise ValueError("Simulated application error")
elif path == '/badrequest':
raise BadRequest("Invalid request format")
# Normal response
status = '200 OK'
headers = [('Content-Type', 'text/plain')]
start_response(status, headers)
return [b'Hello World!']
except BadRequest as e:
# Handle 400-level errors
logger.warning(f"Bad request: {e}")
status = '400 Bad Request'
headers = [('Content-Type', 'text/plain')]
start_response(status, headers)
return [b'Bad Request']
except Exception as e:
# Handle 500-level errors
logger.error(f"Internal error: {e}", exc_info=True)
status = '500 Internal Server Error'
headers = [('Content-Type', 'text/plain')]
start_response(status, headers)
return [b'Internal Server Error']
# Serve with error handling
serve(error_handling_app, host='127.0.0.1', port=8080)Request Size Validation
from waitress import serve
from waitress.utilities import RequestEntityTooLarge, RequestHeaderFieldsTooLarge
def size_validated_app(environ, start_response):
"""Application with manual size validation."""
try:
# Check content length
content_length = int(environ.get('CONTENT_LENGTH', 0))
if content_length > 1024 * 1024: # 1MB limit
raise RequestEntityTooLarge("Request body too large")
# Process request normally
status = '200 OK'
headers = [('Content-Type', 'text/plain')]
start_response(status, headers)
return [b'Request processed successfully']
except RequestEntityTooLarge:
status = '413 Request Entity Too Large'
headers = [('Content-Type', 'text/plain')]
start_response(status, headers)
return [b'Request too large']
# Server configuration with size limits
serve(size_validated_app,
max_request_body_size=2097152, # 2MB server limit
max_request_header_size=65536, # 64KB header limit
host='127.0.0.1',
port=8080
)HTTP Date Processing
Working with HTTP date headers for caching and conditional requests.
from waitress.utilities import build_http_date, parse_http_date
import time
import os
def caching_app(environ, start_response):
"""Application with HTTP date handling for caching."""
# Get file modification time
file_path = '/path/to/resource.txt'
if os.path.exists(file_path):
mtime = os.path.getmtime(file_path)
last_modified = build_http_date(mtime)
# Check If-Modified-Since header
if_modified = environ.get('HTTP_IF_MODIFIED_SINCE')
if if_modified:
try:
client_time = parse_http_date(if_modified)
if client_time >= mtime:
# Not modified
status = '304 Not Modified'
headers = [('Last-Modified', last_modified)]
start_response(status, headers)
return [b'']
except ValueError:
# Invalid date format, ignore
pass
# Return file with Last-Modified header
status = '200 OK'
headers = [
('Content-Type', 'text/plain'),
('Last-Modified', last_modified),
('Cache-Control', 'max-age=3600')
]
start_response(status, headers)
with open(file_path, 'rb') as f:
return [f.read()]
else:
status = '404 Not Found'
headers = [('Content-Type', 'text/plain')]
start_response(status, headers)
return [b'File not found']
# Example date operations
current_time = time.time()
http_date = build_http_date(current_time)
print(f"Current time as HTTP date: {http_date}")
parsed_time = parse_http_date(http_date)
print(f"Parsed back to timestamp: {parsed_time}")Unix Socket Management
Proper Unix socket file handling for production deployments.
from waitress import serve
from waitress.utilities import cleanup_unix_socket
import os
import atexit
def setup_unix_socket_server(app, socket_path):
"""Set up server with proper Unix socket management."""
# Clean up socket file on startup
cleanup_unix_socket(socket_path)
# Register cleanup on exit
atexit.register(cleanup_unix_socket, socket_path)
# Set appropriate permissions before creating socket
umask = os.umask(0o077) # Restrict to owner only initially
try:
serve(app,
unix_socket=socket_path,
unix_socket_perms=0o660, # rw-rw---- (owner and group)
threads=4
)
finally:
# Restore original umask
os.umask(umask)
# Clean up socket file
cleanup_unix_socket(socket_path)
# Usage
def my_app(environ, start_response):
status = '200 OK'
headers = [('Content-Type', 'text/plain')]
start_response(status, headers)
return [b'Unix socket server running!']
setup_unix_socket_server(my_app, '/tmp/myapp.sock')Logging Configuration
Setting up comprehensive logging for production servers.
import logging
import logging.handlers
from waitress import serve
from waitress.utilities import logger, queue_logger
def setup_production_logging():
"""Configure comprehensive logging for production."""
# Main application logger
app_logger = logging.getLogger('myapp')
app_logger.setLevel(logging.INFO)
# Waitress loggers
logger.setLevel(logging.INFO)
queue_logger.setLevel(logging.DEBUG)
# File handler with rotation
file_handler = logging.handlers.RotatingFileHandler(
'/var/log/myapp/waitress.log',
maxBytes=10*1024*1024, # 10MB
backupCount=5
)
# Console handler for immediate feedback
console_handler = logging.StreamHandler()
# Detailed formatter
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
file_handler.setFormatter(formatter)
console_handler.setFormatter(formatter)
# Add handlers to loggers
app_logger.addHandler(file_handler)
app_logger.addHandler(console_handler)
logger.addHandler(file_handler)
queue_logger.addHandler(file_handler)
return app_logger
def logged_app(environ, start_response):
"""Application with comprehensive logging."""
app_logger = logging.getLogger('myapp')
# Log request details
method = environ['REQUEST_METHOD']
path = environ['PATH_INFO']
remote_addr = environ['REMOTE_ADDR']
app_logger.info(f"{remote_addr} - {method} {path}")
try:
# Application logic
status = '200 OK'
headers = [('Content-Type', 'text/plain')]
start_response(status, headers)
return [b'Request processed']
except Exception as e:
app_logger.error(f"Error processing {method} {path}: {e}", exc_info=True)
status = '500 Internal Server Error'
headers = [('Content-Type', 'text/plain')]
start_response(status, headers)
return [b'Internal Server Error']
# Set up logging and serve
production_logger = setup_production_logging()
serve(logged_app,
host='0.0.0.0',
port=8080,
threads=6,
ident='MyApp/1.0' # Custom server identification
)Install with Tessl CLI
npx tessl i tessl/pypi-waitress