a2wsgi

When using a2wsgi middleware, you can access the original protocol-specific metadata within the converted application:

• In an ASGI application (converted from WSGI): Access the original WSGI environ via scope["wsgi_environ"].
• In a WSGI application (converted from ASGI): Access the original ASGI scope via environ["asgi.scope"].

# Inside an ASGI app converted from WSGI
wsgi_env = scope["wsgi_environ"]

# Inside a WSGI app converted from ASGI
asgi_scope = environ["asgi.scope"]

Install the a2wsgi package using pip to enable conversion between WSGI and ASGI applications.

pip install a2wsgi

The WSGIMiddleware is designed specifically for HTTP. It handles other ASGI scope types as follows:

• websocket: If a WebSocket connection is attempted, the middleware automatically sends a websocket.close message with code 1000 and returns.
• lifespan: The middleware handles the lifespan.startup and lifespan.shutdown events by sending the appropriate complete messages back to the server, allowing the ASGI server to proceed with its lifecycle management.

When using WSGIMiddleware, the original ASGI scope is preserved and injected into the WSGI environ dictionary under the key "asgi.scope". This allows you to access ASGI-specific metadata within your WSGI application if needed.

def my_wsgi_app(environ, start_response):
    # Access the original ASGI scope
    scope = environ.get("asgi.scope")
    if scope:
        print(f"Request path: {scope.get('path')}")
    
    start_response('200 OK', [('Content-Type', 'text/plain')])
    return [b"Hello!"]

Use WSGIMiddleware to wrap an existing WSGI application, making it compatible with ASGI servers.

Configuration Options:

• workers: The number of threads in the thread pool used to execute the WSGI application. Defaults to 10.
• send_queue_size: The size of the queue used to direct traffic from the WSGI App to the client. This allows balancing memory usage and responsiveness. Defaults to 10.

Unlike other converters (like those in uvicorn or starlette), a2wsgi does not accumulate request or response content in memory, preventing memory limit issues.

from a2wsgi import WSGIMiddleware

# Basic usage
ASGI_APP = WSGIMiddleware(WSGI_APP)

# Advanced usage with custom workers and queue size
ASGI_APP = WSGIMiddleware(WSGI_APP, workers=15, send_queue_size=15)

The following frameworks are compatible with a2wsgi conversion:

WSGI Frameworks:

• Django (wsgi)
• Kuí (wsgi)
• Falcon (wsgi)
• Pyramid
• Bottle
• Flask

ASGI Frameworks:

• Django (asgi)
• Kuí (asgi)
• Falcon (asgi)
• Starlette
• FastAPI
• Sanic
• Quart

Unsupported:

• aiohttp

When a WSGI application is converted to ASGI, the request body is provided via the wsgi.input key in the environ dictionary. This key points to an instance of the Body class, which implements a file-like interface for reading the asynchronous ASGI request stream synchronously.

Methods

• read(size: int = -1) -> bytes: Reads up to size bytes. If size is -1, reads the entire remaining body.
• readline(limit: int = -1) -> bytes: Reads a single line.
• readlines(hint: int = -1) -> List[bytes]: Reads multiple lines.
• __iter__(): Allows iterating over the body line by line.

def my_wsgi_app(environ, start_response):
    # Get the body from the environ
    body = environ['wsgi.input']
    
    # Read the entire body
    data = body.read()
    
    start_response('200 OK', [('Content-Type', 'text/plain')])
    return [b"Received: " + data]

Use the WSGIMiddleware class to wrap an existing WSGI application, making it compatible with ASGI servers (like Uvicorn or Daphne). This is useful when you have a legacy WSGI app (e.g., Flask, Django) that you want to run in an ASGI environment.

Parameters

• app: The WSGI application to convert.
• workers (optional): The number of worker threads in the ThreadPoolExecutor. Defaults to 10.
• send_queue_size (optional): The size of the internal asyncio.Queue used for sending responses. Defaults to 10.

from a2wsgi import WSGIMiddleware

# Your existing WSGI app
def my_wsgi_app(environ, start_response):
    start_response('200 OK', [('Content-Type', 'text/plain')])
    return [b"Hello, World!"]

# Convert it to ASGI
asgi_app = WSGIMiddleware(my_wsgi_app)

When initializing ASGIMiddleware, you can control the event loop behavior and cleanup timing:

• Custom Event Loop: If your environment requires a specific loop, pass it via the loop argument. If omitted, a new loop is started in a background thread.
• Cleanup Timeout: The wait_time parameter determines how long the middleware waits for the underlying ASGI application's background tasks to complete after the WSGI response has been fully yielded. This prevents the WSGI server from hanging indefinitely if the ASGI app has lingering tasks.

The following events are used in the HTTP lifecycle:

• http.request: Contains the request body and more_body flag.
• http.response.start: Used to send the HTTP status and headers.
• http.response.body: Used to send the response body and more_body flag.
• http.disconnect: Indicates the client disconnected.

class HTTPRequestEvent(TypedDict):
    type: Literal["http.request"]
    body: bytes
    more_body: NotRequired[bool]

class HTTPResponseStartEvent(TypedDict):
    type: Literal["http.response.start"]
    status: int
    headers: NotRequired[Iterable[Tuple[bytes, bytes]]]
    trailers: NotRequired[bool]

class HTTPResponseBodyEvent(TypedDict):
    type: Literal["http.response.body"]
    body: NotRequired[bytes]
    more_body: NotRequired[bool]

class HTTPDisconnectEvent(TypedDict):
    type: Literal["http.disconnect"]