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