ASGI to WSGI

asgi_flask.py

from typing import Callable, Optional

from asgiref.sync import AsyncToSync
from flask import Request, Response

# The HTTP/1.1 spec is a little ambiguous, but defines that headers *should*
# only include ASCII characters. Existing server implementations in python all
# use 'latin-1' for full backward compatibility, but 'ascii' should be a
# sufficient subset for our use cases.
#
# Further reading:
# https://github.com/benoitc/gunicorn/issues/1778
# https://github.com/benoitc/gunicorn/pull/1914
#
# The internal implementation of the GraphQL ASGI application we are using:
# https://github.com/encode/starlette/blob/master/starlette/responses.py#L64
HEADER_ENCODING = 'ascii'

class AsgiState(object):
    response: Optional[Response] = None
    body = b''


def asgiHandleFlaskRequest(asgi_application: Callable, request: Request) -> Response:
    """
    Process a flask request with an ASGI compliant application.

    See: https://github.com/django/asgiref/blob/master/specs/www.rst
    """
    asgiState = AsgiState()

    async def send(event):
        if event['type'] == 'http.response.start':
            asgiState.response = Response(
                status=event['status'],
                headers={
                    key.decode(HEADER_ENCODING): value.decode(HEADER_ENCODING)
                    for key, value in event['headers']
                },
            )
        if event['type'] == 'http.response.body':
            asgiState.body = asgiState.body + event['body']
            if not event.get('more_body', False):
                if asgiState.response is None:
                    raise RuntimeError('received response body, but no response has been started')
                asgiState.response.set_data(asgiState.body)

    async def receive():
        message = {'type': 'http.request', 'body': request.get_data()}
        return message

    client = request.remote_addr.split(':')
    server = request.host.split(':')

    scope = {
        'type': 'http',
        'asgi': {'version': '3.0', 'spec_version': '2.1'},
        'http_version': request.environ.get('SERVER_PROTOCOL'),
        'method': request.method,
        'scheme': request.scheme,
        'path': request.path,
        'query_string': request.query_string,
        'root_path': request.script_root,
        'headers': [
            (key.lower().encode(HEADER_ENCODING), value.encode(HEADER_ENCODING))
            for key, value in request.headers.items()
        ],
        'client': [client[0], int(client[1]) if len(client) > 1 else None],
        'server': [server[0], int(server[1]) if len(server) > 1 else None],
    }

    # Synchronously process ASGI events
    sync_asgi_application = AsyncToSync(asgi_application)
    sync_asgi_application(scope, receive, send)

    if asgiState.response is None:
        raise TypeError('asgi application did not generate a response for request')

    return asgiState.response