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

Overview

Eval results

Files

WSGI Integration

Name: tessl/pypi-asgiref
Author: tessl

Adapters for running WSGI applications within ASGI servers, enabling legacy web applications to benefit from async server capabilities. This allows gradual migration from WSGI to ASGI without requiring complete application rewrites.

Capabilities

WSGI-to-ASGI Adapter

Main adapter class that wraps WSGI applications to run in ASGI servers, handling protocol translation and request/response conversion.

class WsgiToAsgi:
    """Main WSGI-to-ASGI adapter."""
    
    def __init__(self, wsgi_application):
        """
        Initialize adapter with WSGI application.
        
        Parameters:
        - wsgi_application: callable, WSGI application following PEP 3333
        """
    
    def __call__(self, scope, receive, send):
        """
        ASGI application interface.
        
        Parameters:
        - scope: dict, ASGI scope containing request information
        - receive: callable, ASGI receive channel for request body
        - send: callable, ASGI send channel for response
        
        Returns:
        Coroutine that handles the WSGI application execution
        """

Per-Connection Instance

Internal adapter instance that handles individual connections, managing WSGI environ construction and response handling.

class WsgiToAsgiInstance:
    """Per-connection WSGI-to-ASGI adapter instance."""
    
    def __init__(self, wsgi_application):
        """
        Initialize instance with WSGI application.
        
        Parameters:
        - wsgi_application: callable, WSGI application
        """
    
    def __call__(self, scope, receive, send):
        """
        Handle ASGI connection for this instance.
        
        Parameters:
        - scope: dict, ASGI scope
        - receive: callable, ASGI receive channel
        - send: callable, ASGI send channel
        
        Returns:
        Coroutine that processes the request
        """
    
    def build_environ(self, scope, body):
        """
        Build WSGI environ dictionary from ASGI scope.
        
        Parameters:
        - scope: dict, ASGI HTTP scope
        - body: bytes, request body data
        
        Returns:
        dict: WSGI environ dictionary following PEP 3333
        """
    
    def start_response(self, status, response_headers, exc_info=None):
        """
        WSGI start_response callable implementation.
        
        Parameters:
        - status: str, HTTP status (e.g., '200 OK')
        - response_headers: list, list of (name, value) header tuples
        - exc_info: tuple, exception information (optional)
        
        Returns:
        callable: Function to write response body (for compatibility)
        """

Usage Examples

Basic WSGI Application Adaptation

from asgiref.wsgi import WsgiToAsgi

# Simple WSGI application
def simple_wsgi_app(environ, start_response):
    status = '200 OK'
    headers = [('Content-Type', 'text/plain')]
    start_response(status, headers)
    return [b'Hello from WSGI!']

# Convert to ASGI
asgi_app = WsgiToAsgi(simple_wsgi_app)

# Can now be used with ASGI servers
# await asgi_app(scope, receive, send)

Flask Application Integration

from flask import Flask
from asgiref.wsgi import WsgiToAsgi

# Create Flask application
app = Flask(__name__)

@app.route('/')
def hello():
    return 'Hello from Flask via ASGI!'

@app.route('/api/data')
def api_data():
    return {'message': 'API response', 'status': 'success'}

# Convert Flask WSGI app to ASGI
asgi_app = WsgiToAsgi(app.wsgi_app)

# Now compatible with ASGI servers like Uvicorn, Hypercorn, etc.

Django WSGI Integration

import os
import django
from django.core.wsgi import get_wsgi_application
from asgiref.wsgi import WsgiToAsgi

# Configure Django
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
django.setup()

# Get Django WSGI application
django_wsgi_app = get_wsgi_application()

# Convert to ASGI
django_asgi_app = WsgiToAsgi(django_wsgi_app)

# Can be served by ASGI servers alongside native ASGI applications

Mixed ASGI/WSGI Application Router

from asgiref.wsgi import WsgiToAsgi

class MixedRouter:
    """Router that can handle both ASGI and WSGI applications."""
    
    def __init__(self):
        self.routes = []
    
    def add_wsgi_app(self, path_prefix, wsgi_app):
        """Add a WSGI application at the given path prefix."""
        asgi_app = WsgiToAsgi(wsgi_app)
        self.routes.append((path_prefix, asgi_app, 'wsgi'))
    
    def add_asgi_app(self, path_prefix, asgi_app):
        """Add an ASGI application at the given path prefix."""
        self.routes.append((path_prefix, asgi_app, 'asgi'))
    
    async def __call__(self, scope, receive, send):
        path = scope['path']
        
        for prefix, app, app_type in self.routes:
            if path.startswith(prefix):
                # Modify scope to remove prefix
                new_scope = {**scope, 'path': path[len(prefix):]}
                await app(new_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
router = MixedRouter()

# Add WSGI applications (automatically converted)
router.add_wsgi_app('/legacy', simple_wsgi_app)
router.add_wsgi_app('/flask', app.wsgi_app)

# Add native ASGI applications
async def native_asgi_app(scope, receive, send):
    await send({
        'type': 'http.response.start',
        'status': 200,
        'headers': [[b'content-type', b'application/json']],
    })
    await send({
        'type': 'http.response.body',
        'body': b'{"message": "Native ASGI response"}',
    })

router.add_asgi_app('/api', native_asgi_app)

Middleware with WSGI Applications

from asgiref.wsgi import WsgiToAsgi
import time

class TimingMiddleware:
    """Middleware that adds timing headers to responses."""
    
    def __init__(self, app):
        self.app = app
    
    async def __call__(self, scope, receive, send):
        start_time = time.time()
        
        async def send_wrapper(message):
            if message['type'] == 'http.response.start':
                duration = time.time() - start_time
                headers = list(message.get('headers', []))
                headers.append([
                    b'x-response-time', 
                    f'{duration:.3f}s'.encode()
                ])
                message = {**message, 'headers': headers}
            await send(message)
        
        await self.app(scope, receive, send_wrapper)

# Apply middleware to WSGI application
timed_wsgi_app = TimingMiddleware(WsgiToAsgi(simple_wsgi_app))

Custom WSGI Application with Complex Logic

from asgiref.wsgi import WsgiToAsgi
import json
from urllib.parse import parse_qs

def api_wsgi_app(environ, start_response):
    """WSGI application with API-like behavior."""
    method = environ['REQUEST_METHOD']
    path = environ['PATH_INFO']
    
    if path == '/health' and method == 'GET':
        status = '200 OK'
        headers = [('Content-Type', 'application/json')]
        start_response(status, headers)
        return [json.dumps({'status': 'healthy'}).encode()]
    
    elif path == '/echo' and method == 'POST':
        # Read request body
        try:
            content_length = int(environ.get('CONTENT_LENGTH', 0))
        except ValueError:
            content_length = 0
        
        body = environ['wsgi.input'].read(content_length)
        
        status = '200 OK'
        headers = [('Content-Type', 'application/json')]
        start_response(status, headers)
        
        response = {
            'method': method,
            'path': path,
            'body': body.decode('utf-8'),
            'headers': dict(environ.items())
        }
        return [json.dumps(response, default=str).encode()]
    
    else:
        status = '404 Not Found'
        headers = [('Content-Type', 'text/plain')]
        start_response(status, headers)
        return [b'Not Found']

# Convert to ASGI for modern server compatibility
api_asgi_app = WsgiToAsgi(api_wsgi_app)