tessl/pypi-asgiref
ASGI specs, helper code, and adapters for bridging synchronous and asynchronous Python web applications
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
compatibility.mddocs/
ASGI Compatibility
Utilities for ensuring ASGI applications work consistently across different ASGI versions and server implementations. These functions handle the transition from ASGI2 (double-callable) to ASGI3 (single-callable) patterns automatically.
Capabilities
Application Style Detection
Detect whether an ASGI application uses the legacy double-callable pattern or the modern single-callable pattern.
def is_double_callable(application):
"""
Test if an application is legacy-style (double-callable).
Parameters:
- application: callable, ASGI application to test
Returns:
bool: True if application uses double-callable pattern (ASGI2), False otherwise
"""Application Style Conversion
Convert ASGI applications between different callable patterns to ensure compatibility with all ASGI servers.
def double_to_single_callable(application):
"""
Transform double-callable ASGI app to single-callable.
Parameters:
- application: callable, double-callable ASGI2 application
Returns:
callable: Single-callable ASGI3 application
"""
def guarantee_single_callable(application):
"""
Ensure application is in single-callable ASGI3 style.
Parameters:
- application: callable, ASGI application of any style
Returns:
callable: Single-callable ASGI3 application (converted if necessary)
"""Usage Examples
Basic Compatibility Checking
from asgiref.compatibility import is_double_callable, guarantee_single_callable
# ASGI2-style application (legacy double-callable)
def asgi2_app(scope):
async def asgi_coroutine(receive, send):
await send({
'type': 'http.response.start',
'status': 200,
'headers': [[b'content-type', b'text/plain']],
})
await send({
'type': 'http.response.body',
'body': b'Hello from ASGI2',
})
return asgi_coroutine
# ASGI3-style application (modern single-callable)
async def asgi3_app(scope, receive, send):
await send({
'type': 'http.response.start',
'status': 200,
'headers': [[b'content-type', b'text/plain']],
})
await send({
'type': 'http.response.body',
'body': b'Hello from ASGI3',
})
# Check application styles
print(is_double_callable(asgi2_app)) # True
print(is_double_callable(asgi3_app)) # False
# Ensure both are ASGI3 compatible
compatible_app1 = guarantee_single_callable(asgi2_app)
compatible_app2 = guarantee_single_callable(asgi3_app)
print(is_double_callable(compatible_app1)) # False
print(is_double_callable(compatible_app2)) # FalseFramework Integration
from asgiref.compatibility import guarantee_single_callable
class ASGIFramework:
def __init__(self):
self.routes = []
def add_app(self, path, app):
"""Add an application, ensuring ASGI3 compatibility."""
compatible_app = guarantee_single_callable(app)
self.routes.append((path, compatible_app))
async def __call__(self, scope, receive, send):
path = scope['path']
for route_path, app in self.routes:
if path.startswith(route_path):
await app(scope, receive, send)
return
# 404 response
await send({
'type': 'http.response.start',
'status': 404,
'headers': [[b'content-type', b'text/plain']],
})
await send({
'type': 'http.response.body',
'body': b'Not Found',
})
# Usage
framework = ASGIFramework()
# Can add both ASGI2 and ASGI3 apps
framework.add_app('/legacy', asgi2_app)
framework.add_app('/modern', asgi3_app)Middleware Compatibility
from asgiref.compatibility import guarantee_single_callable
class CompatibilityMiddleware:
"""Middleware that ensures all wrapped applications are ASGI3 compatible."""
def __init__(self, app):
self.app = guarantee_single_callable(app)
async def __call__(self, scope, receive, send):
# Add compatibility headers
async def send_wrapper(message):
if message['type'] == 'http.response.start':
headers = list(message.get('headers', []))
headers.append([b'x-asgi-version', b'3.0'])
message = {**message, 'headers': headers}
await send(message)
await self.app(scope, receive, send_wrapper)
# Wrap any ASGI application with compatibility middleware
def create_compatible_app(app):
return CompatibilityMiddleware(app)
# Works with both ASGI2 and ASGI3 applications
compatible_legacy = create_compatible_app(asgi2_app)
compatible_modern = create_compatible_app(asgi3_app)Server Implementation
from asgiref.compatibility import guarantee_single_callable, is_double_callable
import asyncio
class SimpleASGIServer:
def __init__(self, app, host='127.0.0.1', port=8000):
self.app = guarantee_single_callable(app)
self.host = host
self.port = port
async def handle_request(self, reader, writer):
"""Handle a single HTTP request."""
# Read request (simplified)
request_line = await reader.readline()
# Create ASGI scope
scope = {
'type': 'http',
'method': 'GET',
'path': '/',
'headers': [],
'query_string': b'',
}
# ASGI receive callable
async def receive():
return {'type': 'http.request', 'body': b''}
# ASGI send callable
async def send(message):
if message['type'] == 'http.response.start':
status = message['status']
writer.write(f'HTTP/1.1 {status} OK\\r\\n'.encode())
for name, value in message.get('headers', []):
writer.write(f'{name.decode()}: {value.decode()}\\r\\n'.encode())
writer.write(b'\\r\\n')
elif message['type'] == 'http.response.body':
body = message.get('body', b'')
writer.write(body)
writer.close()
# Call ASGI application
await self.app(scope, receive, send)
def run(self):
"""Run the server."""
print(f"Starting server on {self.host}:{self.port}")
# Detect application type for logging
app_type = "ASGI2 (converted)" if is_double_callable(self.app) else "ASGI3"
print(f"Application type: {app_type}")
# Start server (simplified)
loop = asyncio.get_event_loop()
server = loop.run_until_complete(
asyncio.start_server(self.handle_request, self.host, self.port)
)
loop.run_until_complete(server.serve_forever())
# Can serve any ASGI application
server = SimpleASGIServer(asgi2_app) # Automatically converted to ASGI3
# server.run()Install with Tessl CLI
npx tessl i tessl/pypi-asgiref