tessl/pypi-hypercorn
A high-performance ASGI and WSGI web server implementation that provides comprehensive support for modern web protocols including HTTP/1, HTTP/2, WebSockets, and experimental HTTP/3
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
middleware.mddocs/
Middleware
Built-in middleware components for common server functionality including WSGI adaptation, request routing, HTTPS redirection, and proxy header handling. These middleware classes provide reusable components for modifying request and response behavior.
Capabilities
WSGI Middleware for Asyncio
Middleware that enables WSGI application execution within asyncio event loops, handling the synchronous-to-asynchronous adaptation.
class AsyncioWSGIMiddleware:
"""
WSGI middleware adapter for asyncio event loops.
Enables WSGI applications to run within asyncio-based servers
by providing the necessary synchronous execution context and
handling the translation between WSGI's blocking interface
and asyncio's non-blocking event loop.
"""
def __init__(self, wsgi_app: WSGIFramework, max_body_size: int = 16*1024*1024):
"""
Initialize asyncio WSGI middleware.
Args:
wsgi_app: WSGI application to wrap
max_body_size: Maximum request body size in bytes
"""
async def __call__(self, scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable):
"""
Process request through WSGI application.
Args:
scope: ASGI scope dictionary
receive: ASGI receive callable
send: ASGI send callable
"""WSGI Middleware for Trio
Middleware that enables WSGI application execution within trio async frameworks, providing structured concurrency support.
class TrioWSGIMiddleware:
"""
WSGI middleware adapter for trio async framework.
Enables WSGI applications to run within trio-based servers
with support for trio's structured concurrency model and
cancellation semantics.
"""
def __init__(self, wsgi_app: WSGIFramework, max_body_size: int = 16*1024*1024):
"""
Initialize trio WSGI middleware.
Args:
wsgi_app: WSGI application to wrap
max_body_size: Maximum request body size in bytes
"""
async def __call__(self, scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable):
"""
Process request through WSGI application.
Args:
scope: ASGI scope dictionary
receive: ASGI receive callable
send: ASGI send callable
"""Dispatcher Middleware
Middleware for routing requests to different applications based on URL paths, enabling application composition and microservice architectures.
class DispatcherMiddleware:
"""
Route requests to different applications based on path prefixes.
This is an alias for AsyncioDispatcherMiddleware for backward compatibility.
Enables mounting multiple applications at different URL paths,
similar to URL routing but at the application level.
"""
def __init__(self, mounts: dict[str, ASGIFramework]):
"""
Initialize dispatcher middleware.
Args:
mounts: Dictionary mapping path prefixes to ASGI applications
Format: {"/api": api_app, "/admin": admin_app}
"""
async def __call__(self, scope: Scope, receive: Callable, send: Callable):
"""
Route request to appropriate application.
Args:
scope: ASGI scope dictionary (path is examined for routing)
receive: ASGI receive callable
send: ASGI send callable
Routes based on longest matching path prefix.
"""
class AsyncioDispatcherMiddleware:
"""
Asyncio-specific dispatcher middleware for routing requests.
Routes requests to different applications based on path prefixes,
optimized for asyncio event loops.
"""
def __init__(self, mounts: dict[str, ASGIFramework]):
"""
Initialize asyncio dispatcher middleware.
Args:
mounts: Dictionary mapping path prefixes to ASGI applications
"""
async def __call__(self, scope: Scope, receive: Callable, send: Callable):
"""
Route request to appropriate application.
Args:
scope: ASGI scope dictionary
receive: ASGI receive callable
send: ASGI send callable
"""
class TrioDispatcherMiddleware:
"""
Trio-specific dispatcher middleware for routing requests.
Routes requests to different applications based on path prefixes,
optimized for trio async framework with structured concurrency.
"""
def __init__(self, mounts: dict[str, ASGIFramework]):
"""
Initialize trio dispatcher middleware.
Args:
mounts: Dictionary mapping path prefixes to ASGI applications
"""
async def __call__(self, scope: Scope, receive: Callable, send: Callable):
"""
Route request to appropriate application.
Args:
scope: ASGI scope dictionary
receive: ASGI receive callable
send: ASGI send callable
"""HTTP to HTTPS Redirect Middleware
Middleware that automatically redirects HTTP requests to their HTTPS equivalents, enforcing secure connections.
class HTTPToHTTPSRedirectMiddleware:
"""
Redirect HTTP requests to HTTPS.
Automatically redirects all HTTP requests to their HTTPS
equivalents with a 301 permanent redirect. Useful for
enforcing HTTPS-only access to applications.
"""
def __init__(self, app: ASGIFramework | WSGIFramework, host: str | None = None):
"""
Initialize HTTPS redirect middleware.
Args:
app: Application to wrap (receives HTTPS requests)
host: Optional host to redirect to. If None, uses request host
"""
async def __call__(self, scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable):
"""
Process request and redirect HTTP to HTTPS.
Args:
scope: ASGI scope dictionary (scheme is checked)
receive: ASGI receive callable
send: ASGI send callable
HTTP requests receive a 301 redirect to the HTTPS URL.
HTTPS requests are passed through to the wrapped application.
"""Proxy Fix Middleware
Middleware for handling proxy headers to correctly determine client information when running behind reverse proxies or load balancers.
class ProxyFixMiddleware:
"""
Handle proxy headers for client information.
Corrects client IP addresses, schemes, and hosts when running
behind reverse proxies or load balancers by processing standard
proxy headers like X-Forwarded-For, X-Forwarded-Proto, etc.
"""
def __init__(
self,
app: ASGIFramework | WSGIFramework,
mode: Literal["legacy", "modern"] = "legacy",
trusted_hops: int = 1
):
"""
Initialize proxy fix middleware.
Args:
app: Application to wrap
mode: Header parsing mode ("legacy" or "modern")
trusted_hops: Number of proxy hops to trust
The trusted_hops parameter controls how many proxy entries
are trusted to prevent header spoofing from untrusted sources.
"""
async def __call__(self, scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable):
"""
Process request and fix proxy headers.
Args:
scope: ASGI scope dictionary (headers are modified)
receive: ASGI receive callable
send: ASGI send callable
Modifies the scope to reflect the actual client information
based on trusted proxy headers.
"""Usage Examples
WSGI Middleware Usage
from hypercorn.middleware import AsyncioWSGIMiddleware
def wsgi_app(environ, start_response):
status = '200 OK'
headers = [('Content-Type', 'text/plain')]
start_response(status, headers)
return [b'Hello from WSGI middleware']
# Wrap WSGI app for asyncio
middleware = AsyncioWSGIMiddleware(wsgi_app, max_body_size=1024*1024)
# Use with Hypercorn
from hypercorn.config import Config
from hypercorn.asyncio import serve
config = Config()
asyncio.run(serve(middleware, config))Dispatcher Middleware
from hypercorn.middleware import DispatcherMiddleware
# Define different applications
async def api_app(scope, receive, send):
await send({'type': 'http.response.start', 'status': 200})
await send({'type': 'http.response.body', 'body': b'API Response'})
async def admin_app(scope, receive, send):
await send({'type': 'http.response.start', 'status': 200})
await send({'type': 'http.response.body', 'body': b'Admin Panel'})
# Create dispatcher
dispatcher = DispatcherMiddleware({
'/api': api_app,
'/admin': admin_app,
'/': main_app # Default application
})
# Requests to /api/* go to api_app
# Requests to /admin/* go to admin_app
# All other requests go to main_appHTTPS Redirect Middleware
from hypercorn.middleware import HTTPToHTTPSRedirectMiddleware
# Wrap application to enforce HTTPS
https_app = HTTPToHTTPSRedirectMiddleware(app)
# Or redirect to specific host
https_app = HTTPToHTTPSRedirectMiddleware(app, host="secure.example.com")
# HTTP requests get 301 redirect to HTTPS
# HTTPS requests are passed through to appProxy Fix Middleware
from hypercorn.middleware import ProxyFixMiddleware
# Basic proxy fix (trust 1 level of proxies)
proxy_app = ProxyFixMiddleware(app)
# Behind multiple proxies/load balancers
proxy_app = ProxyFixMiddleware(
app,
x_for=2, # Trust last 2 X-Forwarded-For entries
x_proto=1, # Trust last 1 X-Forwarded-Proto entry
x_host=1, # Trust last 1 X-Forwarded-Host entry
x_prefix=1 # Trust last 1 X-Forwarded-Prefix entry
)
# Now scope['client'] reflects actual client IP
# scope['scheme'] reflects actual scheme (http/https)Combining Middleware
from hypercorn.middleware import (
ProxyFixMiddleware,
HTTPToHTTPSRedirectMiddleware,
DispatcherMiddleware
)
# Layer middleware (innermost to outermost)
app = my_asgi_app
# Fix proxy headers first
app = ProxyFixMiddleware(app, x_for=1, x_proto=1)
# Then enforce HTTPS
app = HTTPToHTTPSRedirectMiddleware(app)
# Finally, add routing if needed
apps = {'/api': api_app, '/': app}
final_app = DispatcherMiddleware(apps)
# Use final_app with HypercornCustom Middleware Pattern
class CustomMiddleware:
def __init__(self, app):
self.app = app
async def __call__(self, scope, receive, send):
# Pre-processing
if scope['type'] == 'http':
# Modify request
scope['headers'].append([b'x-custom', b'value'])
# Call wrapped application
await self.app(scope, receive, self.send_wrapper(send))
def send_wrapper(self, send):
async def wrapped_send(message):
# Post-processing
if message['type'] == 'http.response.start':
# Modify response headers
message.setdefault('headers', [])
message['headers'].append([b'x-processed', b'true'])
await send(message)
return wrapped_send
# Usage
wrapped_app = CustomMiddleware(original_app)Install with Tessl CLI
npx tessl i tessl/pypi-hypercorn