(self, request: Request)
| 89 | self.wsgi_errors = wsgi_errors |
| 90 | |
| 91 | def handle_request(self, request: Request) -> Response: |
| 92 | request.read() |
| 93 | wsgi_input = io.BytesIO(request.content) |
| 94 | |
| 95 | port = request.url.port or {"http": 80, "https": 443}[request.url.scheme] |
| 96 | environ = { |
| 97 | "wsgi.version": (1, 0), |
| 98 | "wsgi.url_scheme": request.url.scheme, |
| 99 | "wsgi.input": wsgi_input, |
| 100 | "wsgi.errors": self.wsgi_errors or sys.stderr, |
| 101 | "wsgi.multithread": True, |
| 102 | "wsgi.multiprocess": False, |
| 103 | "wsgi.run_once": False, |
| 104 | "REQUEST_METHOD": request.method, |
| 105 | "SCRIPT_NAME": self.script_name, |
| 106 | "PATH_INFO": request.url.path, |
| 107 | "QUERY_STRING": request.url.query.decode("ascii"), |
| 108 | "SERVER_NAME": request.url.host, |
| 109 | "SERVER_PORT": str(port), |
| 110 | "SERVER_PROTOCOL": "HTTP/1.1", |
| 111 | "REMOTE_ADDR": self.remote_addr, |
| 112 | } |
| 113 | for header_key, header_value in request.headers.raw: |
| 114 | key = header_key.decode("ascii").upper().replace("-", "_") |
| 115 | if key not in ("CONTENT_TYPE", "CONTENT_LENGTH"): |
| 116 | key = "HTTP_" + key |
| 117 | environ[key] = header_value.decode("ascii") |
| 118 | |
| 119 | seen_status = None |
| 120 | seen_response_headers = None |
| 121 | seen_exc_info = None |
| 122 | |
| 123 | def start_response( |
| 124 | status: str, |
| 125 | response_headers: list[tuple[str, str]], |
| 126 | exc_info: OptExcInfo | None = None, |
| 127 | ) -> typing.Callable[[bytes], typing.Any]: |
| 128 | nonlocal seen_status, seen_response_headers, seen_exc_info |
| 129 | seen_status = status |
| 130 | seen_response_headers = response_headers |
| 131 | seen_exc_info = exc_info |
| 132 | return lambda _: None |
| 133 | |
| 134 | result = self.app(environ, start_response) |
| 135 | |
| 136 | stream = WSGIByteStream(result) |
| 137 | |
| 138 | assert seen_status is not None |
| 139 | assert seen_response_headers is not None |
| 140 | if seen_exc_info and seen_exc_info[0] and self.raise_app_exceptions: |
| 141 | raise seen_exc_info[1] |
| 142 | |
| 143 | status_code = int(seen_status.split()[0]) |
| 144 | headers = [ |
| 145 | (key.encode("ascii"), value.encode("ascii")) |
| 146 | for key, value in seen_response_headers |
| 147 | ] |
| 148 |
nothing calls this directly
no test coverage detected