The comprehensive WSGI web application library providing essential utilities and components for building Python web applications.
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
Feature-rich WSGI development server with automatic reloading, interactive debugging, SSL support, and multi-threading capabilities. The development server provides a convenient way to run and test applications during development with extensive debugging and monitoring features.
⚠️ Warning: This server is designed for development only. Never use it in production environments as it is not designed to be stable, secure, or efficient.
The primary function for starting the development server with comprehensive configuration options.
def run_simple(hostname, port, application, use_reloader=False, use_debugger=False, use_evalex=True, extra_files=None, exclude_patterns=None, reloader_interval=1, reloader_type="auto", threaded=False, processes=1, request_handler=None, static_files=None, passthrough_errors=False, ssl_context=None):
"""
Start a development server for a WSGI application.
Parameters:
- hostname: Host to bind to (e.g., 'localhost', '0.0.0.0')
- port: Port number to bind to (use 0 for random free port)
- application: WSGI application callable to serve
- use_reloader: Enable automatic reloading when files change
- use_debugger: Enable interactive debugging on exceptions
- use_evalex: Allow interactive Python console in debugger
- extra_files: Additional files to watch for reloading
- exclude_patterns: File patterns to ignore for reloading (fnmatch format)
- reloader_interval: How often to check for file changes (seconds)
- reloader_type: Type of reloader ('auto', 'stat', 'watchdog')
- threaded: Enable multi-threading for concurrent requests
- processes: Number of processes for handling requests (alternative to threading)
- request_handler: Custom request handler class
- static_files: Dict mapping URL prefixes to directory paths for static files
- passthrough_errors: Don't catch unhandled exceptions at server level
- ssl_context: SSL configuration for HTTPS ('adhoc', SSLContext, or (cert, key) tuple)
"""Lower-level function for creating server instances with more control.
def make_server(host, port, app, threaded=False, processes=1, request_handler=None, passthrough_errors=False, ssl_context=None, fd=None):
"""
Create a WSGI server instance.
This provides more control than run_simple() and returns the server object
for advanced usage like running in separate threads.
Parameters:
- host: Host address to bind to
- port: Port number to bind to
- app: WSGI application to serve
- threaded: Enable threading for concurrent requests
- processes: Number of worker processes
- request_handler: Custom request handler class
- passthrough_errors: Don't catch exceptions at server level
- ssl_context: SSL configuration
- fd: Existing file descriptor to use
Returns:
BaseWSGIServer instance (or subclass based on threading/process options)
"""The server class hierarchy provides different concurrency models.
class BaseWSGIServer(HTTPServer):
def __init__(self, host, port, app, handler=None, passthrough_errors=False, ssl_context=None, fd=None):
"""
Single-threaded WSGI server.
Parameters:
- host: Host address
- port: Port number
- app: WSGI application
- handler: Request handler class
- passthrough_errors: Error handling mode
- ssl_context: SSL configuration
- fd: File descriptor for socket
"""
# Properties
multithread: bool = False # Server supports threading
multiprocess: bool = False # Server supports multiple processes
def serve_forever(self, poll_interval=0.5):
"""
Handle requests until shutdown.
Parameters:
- poll_interval: Time between polling for shutdown
"""
def shutdown_request(self, request):
"""Clean up after handling a request."""
def handle_error(self, request, client_address):
"""Handle errors during request processing."""
class ThreadedWSGIServer(socketserver.ThreadingMixIn, BaseWSGIServer):
def __init__(self, host, port, app, handler=None, passthrough_errors=False, ssl_context=None, fd=None):
"""
Multi-threaded WSGI server for concurrent request handling.
Uses ThreadingMixIn to spawn a new thread for each request.
"""
multithread: bool = True
daemon_threads: bool = True # Threads don't prevent server shutdown
class ForkingWSGIServer(ForkingMixIn, BaseWSGIServer):
def __init__(self, host, port, app, processes, handler=None, passthrough_errors=False, ssl_context=None, fd=None):
"""
Multi-process WSGI server for concurrent request handling.
Forks worker processes to handle requests. Only available on Unix-like systems.
Parameters:
- processes: Maximum number of worker processes
"""
multiprocess: bool = True
max_children: int # Maximum number of child processesCustom request handler for WSGI applications with enhanced features.
class WSGIRequestHandler(BaseHTTPRequestHandler):
def __init__(self, request, client_address, server):
"""
WSGI-compatible HTTP request handler.
Handles the conversion between HTTP and WSGI protocols.
"""
# Properties
server: BaseWSGIServer
environ: dict # WSGI environment
def handle_one_request(self):
"""Handle a single HTTP request and convert to WSGI."""
def run_wsgi(self):
"""Execute the WSGI application and return response."""
def log_request(self, code='-', size='-'):
"""Log successful requests."""
def log_error(self, format, *args):
"""Log error messages."""Functions for SSL/TLS configuration and certificate management.
def generate_adhoc_ssl_context():
"""
Generate a self-signed SSL context for development.
Creates temporary certificates that are valid for localhost.
Requires the cryptography library.
Returns:
ssl.SSLContext configured with self-signed certificate
"""
def load_ssl_context(cert_file, key_file=None, protocol=None):
"""
Load SSL context from certificate files.
Parameters:
- cert_file: Path to certificate file (.pem or .crt)
- key_file: Path to private key file (optional if cert_file contains both)
- protocol: SSL protocol version to use
Returns:
ssl.SSLContext configured with the provided certificates
"""
def make_ssl_devcert(base_path, host="localhost", cn=None):
"""
Create a self-signed development certificate.
Parameters:
- base_path: Base path for certificate files (will create .crt and .key files)
- host: Hostname for certificate
- cn: Common name (defaults to host)
Returns:
Tuple of (cert_file_path, key_file_path)
"""
def generate_adhoc_ssl_pair(cn=None):
"""
Generate a self-signed certificate pair.
Parameters:
- cn: Common name for certificate
Returns:
Tuple of (Certificate, RSAPrivateKey) objects
"""Helper functions for server management and detection.
def is_running_from_reloader():
"""
Check if the current process is running from the Werkzeug reloader.
Returns:
True if running as a reloader subprocess
"""
def select_address_family(host, port):
"""
Automatically select appropriate address family (IPv4/IPv6) for host.
Parameters:
- host: Host address
- port: Port number
Returns:
socket.AddressFamily (AF_INET or AF_INET6)
"""
def get_interface_ip(family):
"""
Get the IP address of the network interface.
Parameters:
- family: Address family (AF_INET or AF_INET6)
Returns:
IP address string
"""from werkzeug.serving import run_simple
from werkzeug.wrappers import Request, Response
def application(environ, start_response):
request = Request(environ)
response = Response(f'Hello from {request.path}!')
return response(environ, start_response)
if __name__ == '__main__':
# Basic server
run_simple('localhost', 8000, application)from werkzeug.serving import run_simple
from werkzeug.debug import DebuggedApplication
def create_app():
def app(environ, start_response):
request = Request(environ)
if request.path == '/error':
# Trigger an error for debugging
raise ValueError("This is a test error!")
response = Response(f'Path: {request.path}')
return response(environ, start_response)
return app
if __name__ == '__main__':
app = create_app()
# Development server with debugging and auto-reload
run_simple(
'localhost',
8000,
app,
use_debugger=True, # Interactive debugger on errors
use_reloader=True, # Auto-reload on file changes
use_evalex=True # Interactive Python console in debugger
)from werkzeug.serving import run_simple
import time
import threading
def slow_application(environ, start_response):
request = Request(environ)
# Simulate slow processing
time.sleep(2)
thread_id = threading.get_ident()
response = Response(f'Processed by thread {thread_id}')
return response(environ, start_response)
if __name__ == '__main__':
# Multi-threaded server for concurrent requests
run_simple(
'localhost',
8000,
slow_application,
threaded=True # Handle multiple requests concurrently
)from werkzeug.serving import run_simple, generate_adhoc_ssl_context
def secure_app(environ, start_response):
request = Request(environ)
# Check if request is secure
is_secure = request.environ.get('wsgi.url_scheme') == 'https'
response = Response(f'Secure: {is_secure}')
return response(environ, start_response)
if __name__ == '__main__':
# Option 1: Self-signed certificate
run_simple(
'localhost',
8000,
secure_app,
ssl_context='adhoc' # Auto-generate self-signed cert
)
# Option 2: Custom certificate files
# run_simple(
# 'localhost',
# 8000,
# secure_app,
# ssl_context=('cert.pem', 'key.pem')
# )from werkzeug.serving import run_simple
def api_app(environ, start_response):
request = Request(environ)
if request.path.startswith('/api/'):
# Handle API requests
response = Response('{"status": "ok"}', mimetype='application/json')
else:
# Let static file middleware handle other requests
response = Response('Not Found', status=404)
return response(environ, start_response)
if __name__ == '__main__':
run_simple(
'localhost',
8000,
api_app,
static_files={
'/': '/path/to/static/files', # Serve static files from root
'/uploads': '/path/to/upload/dir', # Serve uploads from /uploads
'/assets': ('/path/to/assets', 'text/css') # With custom MIME type
}
)from werkzeug.serving import run_simple, make_server
import os
import threading
def advanced_app(environ, start_response):
request = Request(environ)
response = Response(f'Hello from PID {os.getpid()}')
return response(environ, start_response)
def run_with_custom_config():
# Advanced reloader configuration
run_simple(
'0.0.0.0', # Bind to all interfaces
8000,
advanced_app,
use_reloader=True,
reloader_type='watchdog', # Efficient file watching
reloader_interval=0.5, # Check every 0.5 seconds
extra_files=[ # Watch additional files
'config.yaml',
'templates/base.html'
],
exclude_patterns=[ # Ignore certain files
'*.pyc',
'*/__pycache__/*',
'*.log'
]
)
def run_with_server_object():
# Using make_server for more control
server = make_server(
'localhost',
8000,
advanced_app,
threaded=True
)
print(f"Server running on {server.server_address}")
# Run in separate thread
server_thread = threading.Thread(target=server.serve_forever)
server_thread.daemon = True
server_thread.start()
# Do other work...
input("Press Enter to stop server...")
server.shutdown()
server.server_close()
if __name__ == '__main__':
# Choose configuration
run_with_custom_config()
# run_with_server_object()from werkzeug.serving import run_simple, WSGIRequestHandler
from werkzeug.wrappers import Request, Response
import logging
class CustomRequestHandler(WSGIRequestHandler):
def log_request(self, code='-', size='-'):
# Custom request logging
logging.info(f'{self.client_address[0]} - {self.command} {self.path} - {code}')
def log_error(self, format, *args):
# Custom error logging
logging.error(f'{self.client_address[0]} - ERROR: {format % args}')
def logged_app(environ, start_response):
request = Request(environ)
# Log request details
logging.info(f'Processing {request.method} {request.path}')
response = Response('Logged request')
return response(environ, start_response)
if __name__ == '__main__':
# Setup logging
logging.basicConfig(level=logging.INFO)
run_simple(
'localhost',
8000,
logged_app,
request_handler=CustomRequestHandler
)from werkzeug.serving import make_ssl_devcert, load_ssl_context, run_simple
def setup_ssl_server():
# Generate development certificates
cert_path, key_path = make_ssl_devcert('dev-cert', host='localhost')
# Load SSL context
ssl_context = load_ssl_context(cert_path, key_path)
# Or use the files directly
run_simple(
'localhost',
8000,
app,
ssl_context=(cert_path, key_path)
)
def app(environ, start_response):
request = Request(environ)
# SSL information available in environ
ssl_info = {
'scheme': request.scheme,
'is_secure': request.is_secure,
'url': request.url
}
response = Response(f'SSL Info: {ssl_info}')
return response(environ, start_response)
if __name__ == '__main__':
setup_ssl_server()from werkzeug.serving import run_simple
from werkzeug.debug import DebuggedApplication
from werkzeug.exceptions import HTTPException, NotFound
def debug_app(environ, start_response):
request = Request(environ)
if request.path == '/error':
# Unhandled exception - will show in debugger
raise ValueError("This is an unhandled error")
elif request.path == '/http-error':
# HTTP exception - proper error response
raise NotFound("Page not found")
elif request.path == '/debug-info':
# Show debug information
debug_info = {
'path': request.path,
'method': request.method,
'headers': dict(request.headers),
'args': dict(request.args)
}
response = Response(str(debug_info))
else:
response = Response('Hello! Try /error, /http-error, or /debug-info')
return response(environ, start_response)
if __name__ == '__main__':
# Wrap with debugger for development
debugged_app = DebuggedApplication(debug_app, evalex=True)
run_simple(
'localhost',
8000,
debugged_app,
use_debugger=True, # Enable Werkzeug debugger
passthrough_errors=False # Let debugger catch errors
)from werkzeug.serving import run_simple, is_running_from_reloader
import os
def production_ready_app(environ, start_response):
request = Request(environ)
response = Response('Production app')
return response(environ, start_response)
def main():
# Check environment
is_development = os.getenv('FLASK_ENV') == 'development'
if is_development:
# Development configuration
print("Running in development mode")
run_simple(
'localhost',
8000,
production_ready_app,
use_debugger=True,
use_reloader=True,
use_evalex=True
)
else:
# Production reminder
print("⚠️ Don't use run_simple in production!")
print("Use a proper WSGI server like Gunicorn, uWSGI, or mod_wsgi")
# For demonstration only - don't do this in real production
run_simple(
'127.0.0.1', # Only local access
8000,
production_ready_app,
threaded=True,
use_debugger=False,
use_reloader=False
)
if __name__ == '__main__':
main()Install with Tessl CLI
npx tessl i tessl/pypi-werkzeug