Builds a scope and request message into a WSGI environ object.
(scope: HTTPScope, message: ASGIReceiveEvent, body: io.BytesIO)
| 25 | |
| 26 | |
| 27 | def build_environ(scope: HTTPScope, message: ASGIReceiveEvent, body: io.BytesIO) -> Environ: |
| 28 | """ |
| 29 | Builds a scope and request message into a WSGI environ object. |
| 30 | """ |
| 31 | script_name = scope.get("root_path", "").encode("utf8").decode("latin1") |
| 32 | path_info = scope["path"].encode("utf8").decode("latin1") |
| 33 | if path_info.startswith(script_name): |
| 34 | path_info = path_info[len(script_name) :] |
| 35 | environ = { |
| 36 | "REQUEST_METHOD": scope["method"], |
| 37 | "SCRIPT_NAME": script_name, |
| 38 | "PATH_INFO": path_info, |
| 39 | "QUERY_STRING": scope["query_string"].decode("ascii"), |
| 40 | "SERVER_PROTOCOL": "HTTP/%s" % scope["http_version"], |
| 41 | "wsgi.version": (1, 0), |
| 42 | "wsgi.url_scheme": scope.get("scheme", "http"), |
| 43 | "wsgi.input": body, |
| 44 | "wsgi.errors": sys.stdout, |
| 45 | "wsgi.multithread": True, |
| 46 | "wsgi.multiprocess": True, |
| 47 | "wsgi.run_once": False, |
| 48 | } |
| 49 | |
| 50 | # Get server name and port - required in WSGI, not in ASGI |
| 51 | server = scope.get("server") |
| 52 | if server is None: |
| 53 | server = ("localhost", 80) |
| 54 | environ["SERVER_NAME"] = server[0] |
| 55 | environ["SERVER_PORT"] = server[1] |
| 56 | |
| 57 | # Get client IP address |
| 58 | client = scope.get("client") |
| 59 | if client is not None: |
| 60 | environ["REMOTE_ADDR"] = client[0] |
| 61 | |
| 62 | # Go through headers and make them into environ entries |
| 63 | for name, value in scope.get("headers", []): |
| 64 | name_str: str = name.decode("latin1") |
| 65 | if name_str == "content-length": |
| 66 | corrected_name = "CONTENT_LENGTH" |
| 67 | elif name_str == "content-type": |
| 68 | corrected_name = "CONTENT_TYPE" |
| 69 | else: |
| 70 | corrected_name = "HTTP_%s" % name_str.upper().replace("-", "_") |
| 71 | # HTTPbis say only ASCII chars are allowed in headers, but we latin1 |
| 72 | # just in case |
| 73 | value_str: str = value.decode("latin1") |
| 74 | if corrected_name in environ: |
| 75 | corrected_name_environ = environ[corrected_name] |
| 76 | assert isinstance(corrected_name_environ, str) |
| 77 | value_str = corrected_name_environ + "," + value_str |
| 78 | environ[corrected_name] = value_str |
| 79 | return environ |
| 80 | |
| 81 | |
| 82 | class _WSGIMiddleware: |