Function create_wsgi_request

zappa/wsgi.py:18–117

Given some event_info via API Gateway, create and return a valid WSGI request environ.

(
    event_info,
    server_name="zappa",
    script_name=None,
    trailing_slash=True,
    binary_support=False,
    base_path=None,
    context_header_mappings=None,
)

def create_wsgi_request(
    event_info,
    server_name="zappa",
    script_name=None,
    trailing_slash=True,
    binary_support=False,
    base_path=None,
    context_header_mappings=None,
):
    """
    Given some event_info via API Gateway,
    create and return a valid WSGI request environ.
    """
    if event_info.get("version", "") == "2.0":
        method, headers, path, query_string, remote_user, authorizer = process_lambda_payload_v2(event_info)
    else:
        method, headers, path, query_string, remote_user, authorizer = process_lambda_payload_v1(event_info)

    resolve_context_headers(event_info, headers, context_header_mappings)

    # Related:  https://github.com/Miserlou/Zappa/issues/677
    #           https://github.com/Miserlou/Zappa/issues/683
    #           https://github.com/Miserlou/Zappa/issues/696
    #           https://github.com/Miserlou/Zappa/issues/836
    #           https://en.wikipedia.org/wiki/Hypertext_Transfer_Protocol#Summary_table
    body = extract_request_body(event_info, method, binary_support)

    # Make header names canonical, e.g. content-type => Content-Type
    # https://github.com/Miserlou/Zappa/issues/1188
    headers = titlecase_keys(headers)

    if base_path:
        script_name = f"/{base_path}"

        if path.startswith(script_name):
            path = path[len(script_name) :]

    x_forwarded_for = headers.get("X-Forwarded-For", "")
    if "," in x_forwarded_for:
        # The last one is the cloudfront proxy ip. The second to last is the real client ip.
        # Everything else is user supplied and untrustworthy.
        addresses = [addr.strip() for addr in x_forwarded_for.split(",")]
        remote_addr = addresses[-2]
    else:
        remote_addr = x_forwarded_for or "127.0.0.1"

    environ = {
        "PATH_INFO": get_wsgi_string(path),
        "QUERY_STRING": get_wsgi_string(query_string),
        "REMOTE_ADDR": remote_addr,
        "REQUEST_METHOD": method,
        "SCRIPT_NAME": get_wsgi_string(str(script_name)) if script_name else "",
        "SERVER_NAME": str(server_name),
        "SERVER_PORT": headers.get("X-Forwarded-Port", "80"),
        "SERVER_PROTOCOL": str("HTTP/1.1"),
        "wsgi.version": (1, 0),
        "wsgi.url_scheme": headers.get("X-Forwarded-Proto", "http"),
        # This must be Bytes or None